[FF-11969] Add --all flag to confluent kafka acl list to include …

…ACLs with only integer IDs (#2186)

Co-authored-by: Brian Strauch <bstrauch@confluent.io>

internal/kafka/command_acl_list.go

@@ -25,6 +25,7 @@ func (c *aclCommand) newListCommand() *cobra.Command {
 	pcmd.AddEnvironmentFlag(cmd, c.AuthenticatedCLICommand)
 	pcmd.AddServiceAccountFlag(cmd, c.AuthenticatedCLICommand)
 	cmd.Flags().String("principal", "", `Principal for this operation, prefixed with "User:".`)
+	cmd.Flags().Bool("all", false, "Include ACLs for deleted principals with integer IDs.")
 	pcmd.AddOutputFlag(cmd)
 
 	cmd.MarkFlagsMutuallyExclusive("service-account", "principal")

pkg/acl/acl.go

@@ -3,6 +3,7 @@ package acl
 import (
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 
 	"github.com/antihax/optional"
@@ -315,10 +316,23 @@ func CreateAclRequestDataToAclData(data *AclRequestDataWithError) cpkafkarestv3.
 }
 
 func PrintACLsFromKafkaRestResponse(cmd *cobra.Command, acls []cckafkarestv3.AclData) error {
+	all, err := cmd.Flags().GetBool("all")
+	if err != nil {
+		return err
+	}
+
 	list := output.NewList(cmd)
 	for _, acl := range acls {
+		principal := acl.GetPrincipal()
+		if !all {
+			if hasIntegerId, err := principalHasIntegerId(principal); err != nil {
+				return err
+			} else if hasIntegerId {
+				continue
+			}
+		}
 		list.Add(&out{
-			Principal:    acl.GetPrincipal(),
+			Principal:    principal,
 			Permission:   acl.GetPermission(),
 			Operation:    acl.GetOperation(),
 			ResourceType: string(acl.GetResourceType()),
@@ -330,6 +344,18 @@ func PrintACLsFromKafkaRestResponse(cmd *cobra.Command, acls []cckafkarestv3.Acl
 	return list.Print()
 }
 
+func principalHasIntegerId(principal string) (bool, error) {
+	x := strings.Split(principal, ":")
+	if len(x) < 2 {
+		return false, errors.Errorf("unrecognized principal format %s", principal)
+	}
+	suffix := x[1]
+
+	// The principal has a numeric ID
+	_, err := strconv.ParseInt(suffix, 10, 32)
+	return err == nil, nil
+}
+
 func GetCreateAclRequestData(binding *ccstructs.ACLBinding) cckafkarestv3.CreateAclRequestData {
 	data := cckafkarestv3.CreateAclRequestData{
 		Host:         binding.GetEntry().GetHost(),

test/fixtures/output/kafka/acl/list-cloud-all.golden

@@ -0,0 +1,4 @@
+    Principal   | Permission | Operation | Resource Type | Resource Name | Pattern Type  
+----------------+------------+-----------+---------------+---------------+---------------
+  User:012345   | ALLOW      | READ      | TOPIC         | test-topic    | LITERAL       
+  User:sa-12345 | ALLOW      | READ      | TOPIC         | test-topic    | LITERAL       

test/fixtures/output/kafka/acl/list-help.golden

@@ -14,6 +14,7 @@ Flags:
       --environment string        Environment ID.
       --service-account string    Service account ID.
       --principal string          Principal for this operation, prefixed with "User:".
+      --all                       Include ACLs for deleted principals with integer IDs.
   -o, --output string             Specify the output format as "human", "json", or "yaml". (default "human")
 
 Global Flags:

test/kafka_test.go

@@ -89,6 +89,7 @@ func (s *CLITestSuite) TestKafka() {
 		{args: "kafka cluster describe lkc-unknown-type", fixture: "kafka/describe-unknown-cluster-type.golden"},
 
 		{args: "kafka acl list --cluster lkc-acls", fixture: "kafka/acl/list-cloud.golden"},
+		{args: "kafka acl list --cluster lkc-acls --all", fixture: "kafka/acl/list-cloud-all.golden"},
 		{args: "kafka acl list --cluster lkc-acls -o json", fixture: "kafka/acl/list-json-cloud.golden"},
 		{args: "kafka acl list --cluster lkc-acls -o yaml", fixture: "kafka/acl/list-yaml-cloud.golden"},
 		{args: "kafka acl create --cluster lkc-acls --allow --service-account 7272 --operations read,described --topic test-topic", fixture: "kafka/acl/invalid-operation.golden", exitCode: 1},

test/test-server/kafka_rest_router.go

@@ -109,6 +109,9 @@ func handleKafkaRestClusters(t *testing.T) http.HandlerFunc {
 // Handler for: "/kafka/v3/clusters/{cluster}/acls"
 func handleKafkaRestACLs(t *testing.T) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		clusterId := vars["cluster"]
+
 		data := []cckafkarestv3.AclData{{
 			ResourceType: cckafkarestv3.TOPIC,
 			ResourceName: "test-topic",
@@ -118,6 +121,17 @@ func handleKafkaRestACLs(t *testing.T) http.HandlerFunc {
 			Principal:    "User:sa-12345",
 			PatternType:  "LITERAL",
 		}}
+		if clusterId == "lkc-acls" {
+			data = append(data, cckafkarestv3.AclData{
+				ResourceType: cckafkarestv3.TOPIC,
+				ResourceName: "test-topic",
+				Operation:    "READ",
+				Permission:   "ALLOW",
+				Host:         "*",
+				Principal:    "User:012345",
+				PatternType:  "LITERAL",
+			})
+		}
 
 		var res any
 
@@ -128,7 +142,7 @@ func handleKafkaRestACLs(t *testing.T) http.HandlerFunc {
 			w.WriteHeader(http.StatusCreated)
 			res = cckafkarestv3.AclData{}
 		case http.MethodDelete:
-			res = cckafkarestv3.InlineResponse200{Data: data}
+			res = cckafkarestv3.InlineResponse200{Data: []cckafkarestv3.AclData{data[0]}}
 		}
 
 		err := json.NewEncoder(w).Encode(res)