Repository: kroxylicious/kroxylicious (java)

Issue:
[multitenant mvp] tenant group isolation
**Is your feature request related to a problem? Please describe.**

The multi-tenant POC implemented the introduced limited multi tenant support for **topics**.   Topic names are prefixed using a tenant prefix (currently derived from the SNI used that the tenant uses to connect).   For effect is that tenants A and B can have a topic `mytopic` without collision.

This issue concerned with extending the POC to support Kafka groups too.  Each RPC that carries a `groupid` must be intercepted and rewritten: for requests this means appending the prefix, for responses it means removing the prefix.

This [discussion](https://github.com/kroxylicious/design/discussions/17) sets of the list of RPCs that need to mapped for multi-tenancy.  This issue is concern with those RPCs that carry a `groupid`. The scope of this work is limited to non-transaction use-cases, so RPCs such as `AddOffsetsToTxn` are out of scope.

**Describe the solution you'd like**

As tenant A, my application must be able to consume from topic `foo` using groupid `bar`, independently of any other tenant, who happens to chose the same group id.

The consumer API supports two consume gestures: manual assignment (`KafkaConsumer#assign`) and subscription (`KafkaConsumer#subscribe`).  The admin API supports a number of APIs for maintenance of a consumer group (`#listConsumerGroups`, `#describeConsumerGroups`, `#deleteConsumerGroup`, these are all in scope.

**Describe alternatives you've considered**
N/A

**Additional context**

This task is large so it should be delivered incrementally (i.e. several PRs).  An obstacle for getting starting is that in Kafka groups are created by side effect, so several RPCs will need to be implemented before a group is actually created on the broker).

A suggested way to chunk the work:

1. Manual assignment / list groups (RPCs: FindCoordinator, OffsetFetch, OffsetCommit, ListGroups)
2. Subscription (RPCs: JoinGroup, SyncGroup, LeaveGroup, Heartbeat)
3. DescribeGroups (RPCs: DescribeGroup)
4. alterConsumerGroupOffsets and deleteConsumerGroupOffsets (RPCs: OffsetCommit, . OffsetDelete)


Pull request:
Map the GroupId for OffsetFetch and OffsetDelete RPCs for alterConsumerGroupOffsets, deleteConsumerGroupOffsets and onOffsetFetchRequest Admin API calls
Fix #190 

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 20598125..f9bf24fc 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
@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.filter.multitenant;
 
+import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -28,6 +29,8 @@ import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.OffsetCommitRequestData;
 import org.apache.kafka.common.message.OffsetCommitResponseData;
+import org.apache.kafka.common.message.OffsetDeleteRequestData;
+import org.apache.kafka.common.message.OffsetDeleteResponseData;
 import org.apache.kafka.common.message.OffsetFetchRequestData;
 import org.apache.kafka.common.message.OffsetFetchResponseData;
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
@@ -61,6 +64,8 @@ import io.kroxylicious.proxy.filter.MetadataRequestFilter;
 import io.kroxylicious.proxy.filter.MetadataResponseFilter;
 import io.kroxylicious.proxy.filter.OffsetCommitRequestFilter;
 import io.kroxylicious.proxy.filter.OffsetCommitResponseFilter;
+import io.kroxylicious.proxy.filter.OffsetDeleteRequestFilter;
+import io.kroxylicious.proxy.filter.OffsetDeleteResponseFilter;
 import io.kroxylicious.proxy.filter.OffsetFetchRequestFilter;
 import io.kroxylicious.proxy.filter.OffsetFetchResponseFilter;
 import io.kroxylicious.proxy.filter.OffsetForLeaderEpochRequestFilter;
@@ -88,6 +93,7 @@ public class MultiTenantTransformationFilter
         FetchRequestFilter, FetchResponseFilter,
         OffsetFetchRequestFilter, OffsetFetchResponseFilter,
         OffsetCommitRequestFilter, OffsetCommitResponseFilter,
+        OffsetDeleteRequestFilter, OffsetDeleteResponseFilter,
         OffsetForLeaderEpochRequestFilter, OffsetForLeaderEpochResponseFilter,
         FindCoordinatorRequestFilter, FindCoordinatorResponseFilter,
         ListGroupsResponseFilter,
@@ -171,7 +177,8 @@ public class MultiTenantTransformationFilter
         request.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));
+            Optional.ofNullable(requestGroup.topics())
+                    .ifPresent(topics -> topics.forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
         });
 
         context.forwardRequest(request);
@@ -212,6 +219,19 @@ public class MultiTenantTransformationFilter
         context.forwardResponse(response);
     }
 
+    @Override
+    public void onOffsetDeleteRequest(RequestHeaderData header, OffsetDeleteRequestData request, KrpcFilterContext context) {
+        applyTenantPrefix(context, request::groupId, request::setGroupId, false);
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onOffsetDeleteResponse(ResponseHeaderData header, OffsetDeleteResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
     @Override
     public void onFetchRequest(RequestHeaderData data, FetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
@@ -327,5 +347,4 @@ public class MultiTenantTransformationFilter
 
     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 a66ebe04..87bf68c7 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -171,9 +171,10 @@ public class MultiTenantIT {
     public void consumeOne(KafkaCluster cluster) throws Exception {
         String config = getConfig(PROXY_ADDRESS, cluster);
         try (var proxy = startProxy(config)) {
+            var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
             produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
-            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE, false);
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, MY_VALUE, false);
         }
     }
 
@@ -181,14 +182,53 @@ public class MultiTenantIT {
     public void consumeOneAndOffsetCommit(KafkaCluster cluster) throws Exception {
         String config = getConfig(PROXY_ADDRESS, cluster);
         try (var proxy = startProxy(config)) {
+            var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
             produceAndVerify(TENANT1_PROXY_ADDRESS,
-                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "2")));
-            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "1", true);
-            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "2", true);
+                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, "2"), inCaseOfFailure()));
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "1", true);
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "2", true);
         }
     }
 
+    @Test
+    public void alterOffsetCommit(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config); var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+            var groupId = testInfo.getDisplayName();
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            produceAndVerify(TENANT1_PROXY_ADDRESS,
+                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, "2"),
+                            inCaseOfFailure()));
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "1", true);
+            var rememberedOffsets = admin.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get();
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "2", true);
+
+            admin.alterConsumerGroupOffsets(groupId, rememberedOffsets).all().get();
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "2", true);
+        }
+    }
+
+    @Test
+    public void deleteConsumerGroupOffsets(KafkaCluster cluster) throws Exception {
+        String config = getConfig(PROXY_ADDRESS, cluster);
+        try (var proxy = startProxy(config); var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+            var groupId = testInfo.getDisplayName();
+            createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+            produceAndVerify(TENANT1_PROXY_ADDRESS,
+                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), inCaseOfFailure()));
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "1", true);
+
+            admin.deleteConsumerGroupOffsets(groupId, Set.of(new TopicPartition(NEW_TOPIC_1.name(), 0))).all().get();
+            consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, groupId, MY_KEY, "1", true);
+        }
+    }
+
+    @NotNull
+    private static ProducerRecord<String, String> inCaseOfFailure() {
+        return new ProducerRecord<>(TOPIC_1, MY_KEY, "unexpected - should never be consumed");
+    }
+
     @Test
     public void tenantTopicIsolation(KafkaCluster cluster) throws Exception {
         String config = getConfig(PROXY_ADDRESS, cluster);
@@ -316,13 +356,13 @@ public class MultiTenantIT {
         }
     }
 
-    private void consumeAndVerify(String address, String topicName, String expectedKey, String expectedValue, boolean offsetCommit) {
-        consumeAndVerify(address, topicName, new LinkedList<>(List.of(matchesRecord(topicName, expectedKey, expectedValue))), offsetCommit);
+    private void consumeAndVerify(String address, String topicName, String groupId, String expectedKey, String expectedValue, boolean offsetCommit) {
+        consumeAndVerify(address, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, expectedKey, expectedValue))), offsetCommit);
     }
 
-    private void consumeAndVerify(String address, String topicName, Deque<Predicate<ConsumerRecord<String, String>>> expected, boolean offsetCommit) {
+    private void consumeAndVerify(String address, String topicName, String groupId, Deque<Predicate<ConsumerRecord<String, String>>> expected, boolean offsetCommit) {
         try (var consumer = new KafkaConsumer<String, String>(commonConfig(address, Map.of(
-                ConsumerConfig.GROUP_ID_CONFIG, testInfo.getDisplayName(),
+                ConsumerConfig.GROUP_ID_CONFIG, groupId,
                 ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, Boolean.FALSE.toString(),
                 ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE.toString(),
                 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.toString(),
@@ -338,7 +378,8 @@ public class MultiTenantIT {
                 assertThat(records.partitions()).hasSizeGreaterThanOrEqualTo(1);
                 records.forEach(r -> {
                     assertFalse(expected.isEmpty(), String.format("received unexpected record %s", r));
-                    assertThat(r).matches(expected.pop());
+                    var predicate = expected.pop();
+                    assertThat(r).matches(predicate, predicate.toString());
                 });
             }
 
@@ -438,9 +479,17 @@ public class MultiTenantIT {
 
     @NotNull
     private static <K, V> Predicate<ConsumerRecord<K, V>> matchesRecord(final String expectedTopic, final K expectedKey, final V expectedValue) {
-        return item -> {
-            var rec = ((ConsumerRecord<K, V>) item);
-            return Objects.equals(rec.topic(), expectedTopic) && Objects.equals(rec.key(), expectedKey) && Objects.equals(rec.value(), expectedValue);
+        return new Predicate<>() {
+            @Override
+            public boolean test(ConsumerRecord<K, V> item) {
+                var rec = ((ConsumerRecord<K, V>) item);
+                return Objects.equals(rec.topic(), expectedTopic) && Objects.equals(rec.key(), expectedKey) && Objects.equals(rec.value(), expectedValue);
+            }
+
+            @Override
+            public String toString() {
+                return String.format("expected: key %s value %s", expectedKey, expectedValue);
+            }
         };
     }
 
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetDelete.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetDelete.test.yaml
new file mode 100644
index 00000000..b70f8d15
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetDelete.test.yaml
@@ -0,0 +1,35 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: OFFSET_DELETE
+  version: 0
+  request:
+    payload:
+      groupId: mygroup
+      topics:
+        - name: foo
+          partitions: []
+    diff:
+      - op: replace
+        path: "/groupId"
+        value: tenant1-mygroup
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - name: tenant1-foo
+          partitions:
+            - partitionIndex: 0
+              errorCode: 0
+      throttleTimeMs: 0
+      errorCode: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: foo
+  disabled: false
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 81067fca..47b54d02 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
@@ -43,3 +43,17 @@
         path: "/groups/0/topics/0/name"
         value: foo
   disabled: false
+- apiMessageType: OFFSET_FETCH
+  description: offset fetch group without topic
+  version: 8
+  request:
+    payload:
+      groups:
+        - groupId: mygroup
+          topics: null
+      requireStable: false
+    diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: tenant1-mygroup
+  disabled: false