From 077980216cbd83181ed21a29241031c7d22e2480 Mon Sep 17 00:00:00 2001
From: Nikolaos Kostoulas <nkostoulas@gmail.com>
Date: Tue, 22 Mar 2022 10:52:16 +0000
Subject: [PATCH 1/2] feat: add batched CreateACLs func to ClusterAdmin

Add new method to the `ClusterAdmin` interface to allow batch creation
of `ACL`s. Creating `ACL`s one at a time seems to be very slow and for
applications that need to create a lot of these, it takes time (and
might cause downtime).
---
 admin.go      | 30 +++++++++++++++++++++++++++++-
 admin_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 73 insertions(+), 1 deletion(-)

diff --git a/admin.go b/admin.go
index 3c7894b62..529130fc7 100644
--- a/admin.go
+++ b/admin.go
@@ -76,11 +76,17 @@ type ClusterAdmin interface {
 	// The configs for a particular resource are updated automatically.
 	IncrementalAlterConfig(resourceType ConfigResourceType, name string, entries map[string]IncrementalAlterConfigsEntry, validateOnly bool) error
 
+	// Creates an access control list (ACL) which is bound to a specific resource.
+	// This operation is not transactional so it may succeed or fail.
+	// If you attempt to add an ACL that duplicates an existing ACL, no error will be raised, but
+	// no changes will be made. This operation is supported by brokers with version 0.11.0.0 or higher.
+	CreateACL(resource Resource, acl Acl) error
+
 	// Creates access control lists (ACLs) which are bound to specific resources.
 	// This operation is not transactional so it may succeed for some ACLs while fail for others.
 	// If you attempt to add an ACL that duplicates an existing ACL, no error will be raised, but
 	// no changes will be made. This operation is supported by brokers with version 0.11.0.0 or higher.
-	CreateACL(resource Resource, acl Acl) error
+	CreateACLs([]*ResourceAcls) error
 
 	// Lists access control lists (ACLs) according to the supplied filter.
 	// it may take some time for changes made by createAcls or deleteAcls to be reflected in the output of ListAcls
@@ -803,6 +809,28 @@ func (ca *clusterAdmin) CreateACL(resource Resource, acl Acl) error {
 	return err
 }
 
+func (ca *clusterAdmin) CreateACLs(resourceACLs []*ResourceAcls) error {
+	var acls []*AclCreation
+	for _, resourceACL := range resourceACLs {
+		for _, acl := range resourceACL.Acls {
+			acls = append(acls, &AclCreation{resourceACL.Resource, *acl})
+		}
+	}
+	request := &CreateAclsRequest{AclCreations: acls}
+
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 1
+	}
+
+	b, err := ca.Controller()
+	if err != nil {
+		return err
+	}
+
+	_, err = b.CreateAcls(request)
+	return err
+}
+
 func (ca *clusterAdmin) ListAcls(filter AclFilter) ([]ResourceAcls, error) {
 	request := &DescribeAclsRequest{AclFilter: filter}
 
diff --git a/admin_test.go b/admin_test.go
index 4a4055541..268c6b60e 100644
--- a/admin_test.go
+++ b/admin_test.go
@@ -1118,6 +1118,50 @@ func TestClusterAdminCreateAclErrorHandling(t *testing.T) {
 	}
 }
 
+func TestClusterAdminCreateAcls(t *testing.T) {
+	seedBroker := NewMockBroker(t, 1)
+	defer seedBroker.Close()
+
+	seedBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(seedBroker.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
+		"CreateAclsRequest": NewMockCreateAclsResponse(t),
+	})
+
+	config := NewTestConfig()
+	config.Version = V1_0_0_0
+	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rACLs := []*ResourceAcls{
+		{
+			Resource: Resource{ResourceType: AclResourceTopic, ResourceName: "my_topic"},
+			Acls: []*Acl{
+				{Host: "localhost", Operation: AclOperationAlter, PermissionType: AclPermissionAny},
+			},
+		},
+		{
+			Resource: Resource{ResourceType: AclResourceTopic, ResourceName: "your_topic"},
+			Acls: []*Acl{
+				{Host: "localhost", Operation: AclOperationAlter, PermissionType: AclPermissionAny},
+			},
+		},
+	}
+
+	err = admin.CreateACLs(rACLs)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = admin.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestClusterAdminListAcls(t *testing.T) {
 	seedBroker := NewMockBroker(t, 1)
 	defer seedBroker.Close()

From d712e64b87e87d984387b1c148390fc67e079c99 Mon Sep 17 00:00:00 2001
From: Nikos Kostoulas <nkostoulas@gmail.com>
Date: Wed, 27 Apr 2022 09:42:21 +0100
Subject: [PATCH 2/2] Update admin.go

Co-authored-by: Dominic Evans <dominic.evans@uk.ibm.com>
---
 admin.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/admin.go b/admin.go
index 529130fc7..3fa7258b1 100644
--- a/admin.go
+++ b/admin.go
@@ -80,6 +80,7 @@ type ClusterAdmin interface {
 	// This operation is not transactional so it may succeed or fail.
 	// If you attempt to add an ACL that duplicates an existing ACL, no error will be raised, but
 	// no changes will be made. This operation is supported by brokers with version 0.11.0.0 or higher.
+	// Deprecated: Use CreateACLs instead.
 	CreateACL(resource Resource, acl Acl) error
 
 	// Creates access control lists (ACLs) which are bound to specific resources.