Add support for RD_KAFKA_RESOURCE_TRANSACTIONAL_ID in adminapi

Author: Jure Zvelc

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)
 	}