Repository: kroxylicious/kroxylicious
Pull request #209 (2023-03-20 12:29:55)

Multitenant - Group mapping - consumer manual assign use-case
This is the first part of #190 which covers the use case of a consumer using #assign() method.  The desire is to commit this, then build on top of it to implement the other group mapping use-cases (#subscribe) etc.





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 e9fb1d27..0f2bea00 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
@@ -14,6 +14,10 @@ import org.apache.kafka.common.message.DeleteTopicsRequestData;
 import org.apache.kafka.common.message.DeleteTopicsResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.FindCoordinatorRequestData;
+import org.apache.kafka.common.message.FindCoordinatorResponseData;
+import org.apache.kafka.common.message.ListGroupsRequestData;
+import org.apache.kafka.common.message.ListGroupsResponseData;
 import org.apache.kafka.common.message.ListOffsetsRequestData;
 import org.apache.kafka.common.message.ListOffsetsResponseData;
 import org.apache.kafka.common.message.MetadataRequestData;
@@ -37,7 +41,11 @@ import io.kroxylicious.proxy.filter.DeleteTopicsRequestFilter;
 import io.kroxylicious.proxy.filter.DeleteTopicsResponseFilter;
 import io.kroxylicious.proxy.filter.FetchRequestFilter;
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
+import io.kroxylicious.proxy.filter.FindCoordinatorRequestFilter;
+import io.kroxylicious.proxy.filter.FindCoordinatorResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ListGroupsRequestFilter;
+import io.kroxylicious.proxy.filter.ListGroupsResponseFilter;
 import io.kroxylicious.proxy.filter.ListOffsetsRequestFilter;
 import io.kroxylicious.proxy.filter.ListOffsetsResponseFilter;
 import io.kroxylicious.proxy.filter.MetadataRequestFilter;
@@ -70,7 +78,9 @@ public class MultiTenantTransformationFilter
         FetchRequestFilter, FetchResponseFilter,
         OffsetFetchRequestFilter, OffsetFetchResponseFilter,
         OffsetCommitRequestFilter, OffsetCommitResponseFilter,
-        OffsetForLeaderEpochRequestFilter, OffsetForLeaderEpochResponseFilter {
+        OffsetForLeaderEpochRequestFilter, OffsetForLeaderEpochResponseFilter,
+        FindCoordinatorRequestFilter, FindCoordinatorResponseFilter,
+        ListGroupsRequestFilter, ListGroupsResponseFilter {
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
@@ -144,14 +154,21 @@ public class MultiTenantTransformationFilter
     @Override
     public void onOffsetFetchRequest(RequestHeaderData data, OffsetFetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        request.groups().forEach(requestGroup -> requestGroup.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
+        request.groups().forEach(requestGroup -> {
+            applyTenantPrefix(context, requestGroup::groupId, requestGroup::setGroupId, false);
+            requestGroup.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        });
+
         context.forwardRequest(request);
     }
 
     @Override
     public void onOffsetFetchResponse(ResponseHeaderData data, OffsetFetchResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        response.groups().forEach(responseGroup -> responseGroup.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false)));
+        response.groups().forEach(responseGroup -> {
+            removeTenantPrefix(context, responseGroup::groupId, responseGroup::setGroupId, false);
+            responseGroup.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        });
         context.forwardResponse(response);
     }
 
@@ -169,6 +186,7 @@ public class MultiTenantTransformationFilter
 
     @Override
     public void onOffsetCommitRequest(RequestHeaderData data, OffsetCommitRequestData request, KrpcFilterContext context) {
+        applyTenantPrefix(context, request::groupId, request::setGroupId, false);
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(request);
     }
@@ -189,27 +207,59 @@ public class MultiTenantTransformationFilter
     public void onFetchResponse(ResponseHeaderData data, FetchResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
         context.forwardResponse(response);
+    }
 
+    @Override
+    public void onFindCoordinatorRequest(RequestHeaderData header, FindCoordinatorRequestData request, KrpcFilterContext context) {
+        request.setCoordinatorKeys(request.coordinatorKeys().stream().map(key -> applyTenantPrefix(context, key)).toList());
+        context.forwardRequest(request);
     }
 
-    private void applyTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
+    @Override
+    public void onFindCoordinatorResponse(ResponseHeaderData header, FindCoordinatorResponseData response, KrpcFilterContext context) {
+        response.coordinators().forEach(coordinator -> removeTenantPrefix(context, coordinator::key, coordinator::setKey, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onListGroupsRequest(RequestHeaderData header, ListGroupsRequestData request, KrpcFilterContext context) {
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onListGroupsResponse(ResponseHeaderData header, ListGroupsResponseData response, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
+        var filteredGroups = response.groups().stream().filter(listedGroup -> listedGroup.groupId().startsWith(tenantPrefix)).toList();
+        filteredGroups.forEach(listedGroup -> removeTenantPrefix(context, listedGroup::groupId, listedGroup::setGroupId, false));
+        response.setGroups(filteredGroups);
+        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())) {
             return;
         }
-        setter.accept(tenantPrefix + "-" + clientSideName);
+        setter.accept(applyTenantPrefix(context, clientSideName));
     }
 
-    private void removeTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
+    private String applyTenantPrefix(KrpcFilterContext context, String clientSideName) {
         var tenantPrefix = getTenantPrefix(context);
-        var brokerSideName = getter.get();
+        return tenantPrefix + clientSideName;
+    }
 
+    private void removeTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
+        var brokerSideName = getter.get();
         if (ignoreEmpty && (brokerSideName == null || brokerSideName.isEmpty())) {
             return;
         }
 
-        setter.accept(brokerSideName.substring(tenantPrefix.length() + 1));
+        setter.accept(removeTenantPrefix(context, brokerSideName));
+    }
+
+    private String removeTenantPrefix(KrpcFilterContext context, String brokerSideName) {
+        var tenantPrefix = getTenantPrefix(context);
+        return brokerSideName.substring(tenantPrefix.length());
     }
 
     private static String getTenantPrefix(KrpcFilterContext context) {
@@ -222,9 +272,10 @@ public class MultiTenantTransformationFilter
         if (dot < 1) {
             throw new IllegalStateException("Unexpected SNI hostname formation. SNI hostname : " + sniHostname);
         }
-        return sniHostname.substring(0, dot);
+        return sniHostname.substring(0, dot) + "-";
     }
 
     public MultiTenantTransformationFilter() {
     }
+
 }


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 fe42382b..256976de 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -14,12 +14,15 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.DeleteTopicsResult;
 import org.apache.kafka.clients.admin.ListTopicsResult;
@@ -176,7 +179,7 @@ public class MultiTenantIT {
     }
 
     @Test
-    public void tenantIsolation(KafkaCluster cluster) throws Exception {
+    public void tenantTopicIsolation(KafkaCluster cluster) throws Exception {
         String config = getConfig(PROXY_ADDRESS, cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
@@ -187,6 +190,52 @@ public class MultiTenantIT {
         }
     }
 
+    @Test
+    public void tenantConsumeWithGroup(KafkaCluster cluster) throws Exception {
+        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);
+            verifyConsumerGroups(TENANT1_PROXY_ADDRESS, "Tenant1Group");
+        }
+    }
+
+    @Test
+    public void tenantGroupIsolation(KafkaCluster cluster) throws Exception {
+        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);
+
+            createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            createConsumerWithGroup(TENANT2_PROXY_ADDRESS, "Tenant2Group", NEW_TOPIC_1);
+            verifyConsumerGroups(TENANT2_PROXY_ADDRESS, "Tenant2Group");
+        }
+    }
+
+    private void createConsumerWithGroup(String proxyAddress, String groupId, NewTopic topic) {
+        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(),
+                ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.TRUE.toString(),
+                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.toString(),
+                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)))) {
+
+            var topicPartitions = List.of(new TopicPartition(topic.name(), 0));
+            consumer.assign(topicPartitions);
+
+            consumer.poll(Duration.ofSeconds(1));
+        }
+    }
+
+    private void verifyConsumerGroups(String proxyAddress, String... expectedGroupIds) throws InterruptedException, ExecutionException {
+        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));
+        }
+    }
+
     private void verifyTenant(String address, String... expectedTopics) throws Exception {
         try (var admin = Admin.create(commonConfig(address, Map.of()))) {
 
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/FindCoordinator.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/FindCoordinator.test.yaml
new file mode 100644
index 00000000..f31c8ee9
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/FindCoordinator.test.yaml
@@ -0,0 +1,34 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: FIND_COORDINATOR
+  version: 4
+  request:
+    payload:
+      keyType: 0
+      coordinatorKeys:
+        - test-group
+      timeoutMs: 0
+      validateOnly: false
+    diff:
+      - op: replace
+        path: /coordinatorKeys/0
+        value: tenant1-test-group
+  response:
+    payload:
+      throttleTimeMs: 0
+      coordinators:
+        - key: tenant1-test-group
+          nodeId: 0
+          host: localhost
+          port: 12345
+          errorCode: 0
+          errorMessage:
+    diff:
+      - op: replace
+        path: "/coordinators/0/key"
+        value: test-group
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListGroups.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListGroups.test.yaml
new file mode 100644
index 00000000..8a61579a
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListGroups.test.yaml
@@ -0,0 +1,46 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: LIST_GROUPS
+  description: Group mapping
+  version: 4
+  request:
+    payload:
+      statesFilter: []
+    diff: []
+  response:
+    payload:
+      throttleTimeMs: 0
+      errorCode: 0
+      groups:
+        - groupId: tenant1-mygroup
+          protocolType: consumer
+          groupState: stable
+    diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: mygroup
+  disabled: false
+
+- apiMessageType: LIST_GROUPS
+  description: Group filtering
+  version: 4
+  request:
+    payload:
+      statesFilter: [ ]
+    diff: [ ]
+  response:
+    payload:
+      throttleTimeMs: 0
+      errorCode: 0
+      groups:
+        - groupId: tenant2-mygroup
+          protocolType: consumer
+          groupState: stable
+    diff:
+      - op: remove
+        path: "/groups/0"
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml
index 0c271af1..ab52b7e1 100644
--- a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml
@@ -20,6 +20,9 @@
               committedLeaderEpoch: 0
               committedMetadata: null
     diff:
+      - op: replace
+        path: "/groupId"
+        value: tenant1-mygroup
       - op: replace
         path: "/topics/0/name"
         value: tenant1-foo
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml
index ebd1b78f..81067fca 100644
--- a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml
@@ -15,13 +15,16 @@
               partitionIndexes: [0]
       requireStable: false
     diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: tenant1-mygroup
       - op: replace
         path: "/groups/0/topics/0/name"
         value: tenant1-foo
   response:
     payload:
       groups:
-        - groupId: mygroup
+        - groupId: tenant1-mygroup
           topics:
             - name: tenant1-foo
               partitions:
@@ -33,6 +36,9 @@
           errorCode: 0
       throttleTimeMs: 0
     diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: mygroup
       - op: replace
         path: "/groups/0/topics/0/name"
         value: foo