Add support for RD_KAFKA_RESOURCE_TRANSACTIONAL_ID in adminapi

Author: Jure Zvelc

examples/docker_aws_lambda_example/go.sum

@@ -53,6 +53,7 @@ github.com/confluentinc/confluent-kafka-go/v2 v2.4.0 h1:NbOku86JJlsRJPJKE0snNsz6
 github.com/confluentinc/confluent-kafka-go/v2 v2.4.0/go.mod h1:E1dEQy50ZLfqs7T9luxz0rLxaeFZJZE92XvApJOr/Rk=
 github.com/confluentinc/confluent-kafka-go/v2 v2.5.0/go.mod h1:Hyo+IIQ/tmsfkOcRP8T6VlSeOW3T33v0Me8Xvq4u90Y=
 github.com/confluentinc/confluent-kafka-go/v2 v2.5.3/go.mod h1:QxYLPRKR1MVlkXCCjzjjrpXb0VyFNfVaZXi0obZykJ0=
+github.com/confluentinc/confluent-kafka-go/v2 v2.8.0/go.mod h1:hScqtFIGUI1wqHIgM3mjoqEou4VweGGGX7dMpcUKves=
 github.com/containerd/console v1.0.3 h1:lIr7SlA5PxZyMV30bDW0MGbiOPXwc63yRuCP0ARubLw=
 github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U=
 github.com/containerd/containerd v1.7.12 h1:+KQsnv4VnzyxWcfO9mlxxELaoztsDEjOuCMPAuPqgU0=

kafka/adminapi.go

@@ -493,6 +493,8 @@ const (
 	ResourceUnknown ResourceType = C.RD_KAFKA_RESOURCE_UNKNOWN
 	// ResourceAny - match any resource type (DescribeConfigs)
 	ResourceAny ResourceType = C.RD_KAFKA_RESOURCE_ANY
+	// ResourceTransactionalID - Transactional ID
+	ResourceTransactionalID ResourceType = C.RD_KAFKA_RESOURCE_TRANSACTIONAL_ID
 	// ResourceTopic - Topic
 	ResourceTopic ResourceType = C.RD_KAFKA_RESOURCE_TOPIC
 	// ResourceGroup - Group
@@ -512,6 +514,8 @@ func ResourceTypeFromString(typeString string) (ResourceType, error) {
 	switch strings.ToUpper(typeString) {
 	case "ANY":
 		return ResourceAny, nil
+	case "TRANSACTIONAL_ID":
+		return ResourceTransactionalID, nil
 	case "TOPIC":
 		return ResourceTopic, nil
 	case "GROUP":

kafka/integration_test.go

@@ -2143,6 +2143,15 @@ func (its *IntegrationTestSuite) TestAdminACLs() {
 			Operation:           ACLOperationRead,
 			PermissionType:      ACLPermissionTypeAllow,
 		},
+		{
+			Type:                ResourceTransactionalID,
+			Name:                group,
+			ResourcePatternType: ResourcePatternTypePrefixed,
+			Principal:           "User:test-user-2",
+			Host:                "*",
+			Operation:           ACLOperationAll,
+			PermissionType:      ACLPermissionTypeAllow,
+		},
 		{
 			Type:                ResourceTopic,
 			Name:                topic,
@@ -2213,7 +2222,7 @@ func (its *IntegrationTestSuite) TestAdminACLs() {
 		if err != nil {
 			t.Fatalf("CreateACLs() failed: %s", err)
 		}
-		expectedCreateACLs = []CreateACLResult{{Error: noError}, {Error: noError}, {Error: noError}}
+		expectedCreateACLs = []CreateACLResult{{Error: noError}, {Error: noError}, {Error: noError}, {Error: noError}}
 		checkExpectedResult(expectedCreateACLs, resultCreateACLs)
 	}