Add support for DelegationToken

kafka/kafka_acls.go

@@ -264,40 +264,18 @@ func (c *Client) CreateACL(s StringlyTypedACL) error {
 	return nil
 }
 
-func stringToACLResource(in string) sarama.AclResourceType {
-	switch in {
-	case "Unknown":
-		return sarama.AclResourceUnknown
-	case "Any":
-		return sarama.AclResourceAny
-	case "Topic":
-		return sarama.AclResourceTopic
-	case "Group":
-		return sarama.AclResourceGroup
-	case "Cluster":
-		return sarama.AclResourceCluster
-	case "TransactionalID":
-		return sarama.AclResourceTransactionalID
+func stringToACLResource(in string) (out sarama.AclResourceType) {
+	if err := out.UnmarshalText([]byte(in)); err == nil && out.String() == in { // Forces case-sensitive comparison
+		return
 	}
 	return unknownConversion
 }
 
 func ACLResourceToString(in sarama.AclResourceType) string {
-	switch in {
-	case sarama.AclResourceUnknown:
-		return "Unknown"
-	case sarama.AclResourceAny:
-		return "Any"
-	case sarama.AclResourceTopic:
-		return "Topic"
-	case sarama.AclResourceGroup:
-		return "Group"
-	case sarama.AclResourceCluster:
-		return "Cluster"
-	case sarama.AclResourceTransactionalID:
-		return "TransactionalID"
+	if in == sarama.AclResourceUnknown {
+		return "unknownConversion"
 	}
-	return "unknownConversion"
+	return in.String()
 }
 
 func stringToOperation(in string) sarama.AclOperation {
@@ -456,34 +434,7 @@ func (c *Client) ListACLs() ([]*sarama.ResourceAcls, error) {
 		&sarama.DescribeAclsRequest{
 			Version: int(c.getDescribeAclsRequestAPIVersion()),
 			AclFilter: sarama.AclFilter{
-				ResourceType:              sarama.AclResourceTopic,
-				ResourcePatternTypeFilter: sarama.AclPatternAny,
-				PermissionType:            sarama.AclPermissionAny,
-				Operation:                 sarama.AclOperationAny,
-			},
-		},
-		&sarama.DescribeAclsRequest{
-			Version: int(c.getDescribeAclsRequestAPIVersion()),
-			AclFilter: sarama.AclFilter{
-				ResourceType:              sarama.AclResourceGroup,
-				ResourcePatternTypeFilter: sarama.AclPatternAny,
-				PermissionType:            sarama.AclPermissionAny,
-				Operation:                 sarama.AclOperationAny,
-			},
-		},
-		&sarama.DescribeAclsRequest{
-			Version: int(c.getDescribeAclsRequestAPIVersion()),
-			AclFilter: sarama.AclFilter{
-				ResourceType:              sarama.AclResourceCluster,
-				ResourcePatternTypeFilter: sarama.AclPatternAny,
-				PermissionType:            sarama.AclPermissionAny,
-				Operation:                 sarama.AclOperationAny,
-			},
-		},
-		&sarama.DescribeAclsRequest{
-			Version: int(c.getDescribeAclsRequestAPIVersion()),
-			AclFilter: sarama.AclFilter{
-				ResourceType:              sarama.AclResourceTransactionalID,
+				ResourceType:              sarama.AclResourceAny,
 				ResourcePatternTypeFilter: sarama.AclPatternAny,
 				PermissionType:            sarama.AclPermissionAny,
 				Operation:                 sarama.AclOperationAny,