Repository: kroxylicious/kroxylicious
Pull request #213 (2023-03-21 17:59:29)

Add mapping & integration test for DescribeGroups RPC 

racheljpg: It's only the last commit that needs to be reviewed as the other will be reviewed as part of #212!
k-wall: @SamBarker would you mind re-reviewing?

Patch:
diff --git a/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index b9a903fe..af33cf04 100644
--- a/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -12,6 +12,8 @@ import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.DeleteTopicsRequestData;
 import org.apache.kafka.common.message.DeleteTopicsResponseData;
+import org.apache.kafka.common.message.DescribeGroupsRequestData;
+import org.apache.kafka.common.message.DescribeGroupsResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.FindCoordinatorRequestData;
@@ -47,6 +49,8 @@ import io.kroxylicious.proxy.filter.CreateTopicsRequestFilter;
 import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
 import io.kroxylicious.proxy.filter.DeleteTopicsRequestFilter;
 import io.kroxylicious.proxy.filter.DeleteTopicsResponseFilter;
+import io.kroxylicious.proxy.filter.DescribeGroupsRequestFilter;
+import io.kroxylicious.proxy.filter.DescribeGroupsResponseFilter;
 import io.kroxylicious.proxy.filter.FetchRequestFilter;
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.FindCoordinatorRequestFilter;
@@ -100,7 +104,8 @@ public class MultiTenantTransformationFilter
         JoinGroupRequestFilter,
         SyncGroupRequestFilter,
         LeaveGroupRequestFilter,
-        HeartbeatRequestFilter {
+        HeartbeatRequestFilter,
+        DescribeGroupsRequestFilter, DescribeGroupsResponseFilter {
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
@@ -278,6 +283,18 @@ public class MultiTenantTransformationFilter
         context.forwardRequest(request);
     }
 
+    @Override
+    public void onDescribeGroupsRequest(RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
+        request.setGroups(request.groups().stream().map(group -> applyTenantPrefix(context, group)).toList());
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onDescribeGroupsResponse(ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
+        response.groups().forEach(group -> removeTenantPrefix(context, group::groupId, group::setGroupId, false));
+        context.forwardResponse(response);
+    }
+
     private void applyTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
         String clientSideName = getter.get();
         if (ignoreEmpty && (clientSideName == null || clientSideName.isEmpty())) {


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index 3e0f3f0a..936f6b51 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -22,6 +22,7 @@ import java.util.stream.Stream;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.DeleteTopicsResult;
@@ -38,6 +39,7 @@ import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.ConsumerGroupState;
 import org.apache.kafka.common.TopicCollection;
 import org.apache.kafka.common.TopicCollection.TopicNameCollection;
 import org.apache.kafka.common.TopicPartition;
@@ -205,8 +207,8 @@ public class MultiTenantIT {
         String config = getConfig(PROXY_ADDRESS, cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-            createConsumerWithGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
-            verifyConsumerGroups(TENANT1_PROXY_ADDRESS, "Tenant1Group");
+            runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
+            verifyConsumerGroupsWithList(TENANT1_PROXY_ADDRESS, Set.of("Tenant1Group"));
         }
     }
 
@@ -216,15 +218,51 @@ public class MultiTenantIT {
         String config = getConfig(PROXY_ADDRESS, cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-            createConsumerWithGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
+            runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
 
             createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
-            createConsumerWithGroup(TENANT2_PROXY_ADDRESS, "Tenant2Group", NEW_TOPIC_1, consumerStyle);
-            verifyConsumerGroups(TENANT2_PROXY_ADDRESS, "Tenant2Group");
+            runConsumerInOrderToCreateGroup(TENANT2_PROXY_ADDRESS, "Tenant2Group", NEW_TOPIC_1, consumerStyle);
+            verifyConsumerGroupsWithList(TENANT2_PROXY_ADDRESS, Set.of("Tenant2Group"));
         }
     }
 
-    private void createConsumerWithGroup(String proxyAddress, String groupId, NewTopic topic, ConsumerStyle consumerStyle) {
+    @Test
+    public void describeGroup(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config)) {
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, ConsumerStyle.ASSIGN);
+
+            createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            runConsumerInOrderToCreateGroup(TENANT2_PROXY_ADDRESS, "Tenant2Group", NEW_TOPIC_1, ConsumerStyle.ASSIGN);
+
+            verifyConsumerGroupsWithDescribe(TENANT1_PROXY_ADDRESS, Set.of("Tenant1Group"), Set.of("Tenant2Group", "idontexist"));
+            verifyConsumerGroupsWithDescribe(TENANT2_PROXY_ADDRESS, Set.of("Tenant2Group"), Set.of("Tenant1Group", "idontexist"));
+        }
+    }
+
+    private void verifyConsumerGroupsWithDescribe(String proxyAddress, Set<String> expectedPresent, Set<String> expectedAbsent) throws Exception {
+        try (var admin = Admin.create(commonConfig(proxyAddress, Map.of()))) {
+            var describedGroups = admin.describeConsumerGroups(Stream.concat(expectedPresent.stream(), expectedAbsent.stream()).toList()).all().get();
+            assertThat(describedGroups).hasSize(expectedAbsent.size() + expectedPresent.size());
+
+            var actualPresent = retainKeySubset(describedGroups, expectedPresent);
+            var actualAbsent = retainKeySubset(describedGroups, expectedAbsent);
+
+            // The consumer group comes back as "dead" when it doesn't exist, so we have to check that it's not dead/it is dead if we don't expect to see it
+            assertThat(actualPresent).allSatisfy((s, consumerGroupDescription) -> assertThat(consumerGroupDescription.state()).isNotIn(ConsumerGroupState.DEAD));
+            assertThat(actualAbsent).allSatisfy((s, consumerGroupDescription) -> assertThat(consumerGroupDescription.state()).isIn(ConsumerGroupState.DEAD));
+        }
+    }
+
+    @NotNull
+    private static Map<String, ConsumerGroupDescription> retainKeySubset(Map<String, ConsumerGroupDescription> groups, Set<String> retainedKeys) {
+        var copy = new HashMap<>(groups);
+        copy.keySet().retainAll(retainedKeys);
+        return copy;
+    }
+
+    private void runConsumerInOrderToCreateGroup(String proxyAddress, String groupId, NewTopic topic, ConsumerStyle consumerStyle) {
         try (var consumer = new KafkaConsumer<String, String>(commonConfig(proxyAddress, Map.of(
                 ConsumerConfig.GROUP_ID_CONFIG, groupId,
                 ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, Boolean.FALSE.toString(),
@@ -245,10 +283,10 @@ public class MultiTenantIT {
         }
     }
 
-    private void verifyConsumerGroups(String proxyAddress, String... expectedGroupIds) throws InterruptedException, ExecutionException {
+    private void verifyConsumerGroupsWithList(String proxyAddress, Set<String> expectedGroupIds) throws Exception {
         try (var admin = Admin.create(commonConfig(proxyAddress, Map.of()))) {
             var groups = admin.listConsumerGroups().all().get().stream().map(ConsumerGroupListing::groupId).toList();
-            assertThat(groups).containsExactlyInAnyOrderElementsOf(Set.of(expectedGroupIds));
+            assertThat(groups).containsExactlyInAnyOrderElementsOf(expectedGroupIds);
         }
     }
 
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DescribeGroups.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DescribeGroups.test.yaml
new file mode 100644
index 00000000..997c1922
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DescribeGroups.test.yaml
@@ -0,0 +1,32 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: DESCRIBE_GROUPS
+  version: 5
+  request:
+    payload:
+      groups: [mygroup]
+      includeAuthorizedOperations:  false
+    diff:
+      - op: replace
+        path: "/groups/0"
+        value: tenant1-mygroup
+  response:
+    payload:
+      throttleTimeMs: 0
+      groups:
+        - groupId: tenant1-mygroup
+          errorCode: 0
+          groupState: stable
+          protocolType: consumer
+          protocolData: protocoldata
+          members: []
+          authorizedOperations: 0
+    diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: mygroup
+  disabled: false