Repository: kroxylicious/kroxylicious (java)

Issue:
Custom tenant topic name prefix

**Is your feature request related to a problem? Please describe.**
Current the topic prefix applied seems to be hardcoded, in the underlying kafka cluster as <virt-cluster-name>-

e.g. if topic is foo.bar, and cluster name is john - underlying topic becomes john-foo.bar , because normally seperators are "." or "_" for kafka the "-" can be problematic.

**Describe the solution you'd like**
It would be great if in configuration we could simply define the topic prefix per virtual cluster as a string e.g. "sid_" or "george." so that then this string was applied , this way giving end user control on the prefix and any special character or seperators being used.

**Describe alternatives you've considered**
Living with the "-" and try fix all the other bits to work with "-" seperator :( 


**Additional context**
It looks like this maybe relatively easy to change, by making this non-static, or adding into the filter context another property a field to be set, if set use, else keep existing logic.

Pull request:
Fix #1162: allow tenant / resource name prefix separator to be controlled from configuration
### Type of change

_Select the type of your PR_

- Enhancement / new feature

### Description

With this change, the MultiTenantFilter accepts configuration that allows the separator used to separate the tenant prefix
from the kafka resource name to controlled. If the separator is not specified, the default `-` is used.

```yaml
    filters:
    - type: MultiTenantTransformationFilterFactory
      config:
        prefixResourceNameSeparator: "."
```

Currently, there's no validation that the separator character chosen is legal for Kafka.

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d9c680ff..01758a64 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,7 @@ Please enumerate **all user-facing** changes using format `<github issue/pr numb
 
 ## SNAPSHOT
  
+* [#1162](https://github.com/kroxylicious/kroxylicious/issues/1162): Fix #1162: allow tenant / resource name prefix separator to be controlled from configuration
 * [#1120](https://github.com/kroxylicious/kroxylicious/pull/1120): Generate API compatability report as part of the release process.
  
 ## 0.5.1
diff --git a/kroxylicious-filters/kroxylicious-multitenant/pom.xml b/kroxylicious-filters/kroxylicious-multitenant/pom.xml
index 5e2758ab..12b94eef 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-filters/kroxylicious-multitenant/pom.xml
@@ -22,8 +22,7 @@
     <packaging>jar</packaging>
 
     <name>Multitenancy filter</name>
-    <description>A filter to implement transparent multitenancy for Apache Kafka
-    </description>
+    <description>A filter to implement transparent multitenancy for Apache Kafka</description>
 
     <dependencies>
         <!-- project dependencies - runtime and compile -->
@@ -31,6 +30,10 @@
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.github.spotbugs</groupId>
+            <artifactId>spotbugs-annotations</artifactId>
+        </dependency>
 
         <!-- project dependencies - test -->
         <dependency>
@@ -40,14 +43,14 @@
         </dependency>
 
         <!-- third party dependencies - runtime and compile -->
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+        </dependency>
 
         <!-- third party dependencies - test -->
         <dependency>
@@ -60,11 +63,6 @@
             <artifactId>guava</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-slf4j2-impl</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index 7684fef1..95a68012 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -5,11 +5,13 @@
  */
 package io.kroxylicious.proxy.filter.multitenant;
 
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData;
 import org.apache.kafka.common.message.AddPartitionsToTxnResponseData;
@@ -51,8 +53,6 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.message.SyncGroupRequestData;
 import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
 import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.filter.AddOffsetsToTxnRequestFilter;
 import io.kroxylicious.proxy.filter.AddPartitionsToTxnRequestFilter;
@@ -96,6 +96,9 @@ import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.filter.SyncGroupRequestFilter;
 import io.kroxylicious.proxy.filter.TxnOffsetCommitRequestFilter;
 import io.kroxylicious.proxy.filter.TxnOffsetCommitResponseFilter;
+import io.kroxylicious.proxy.filter.multitenant.config.MultiTenantConfig;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
 
 /**
  * Simple multi-tenant filter.
@@ -104,7 +107,6 @@ import io.kroxylicious.proxy.filter.TxnOffsetCommitResponseFilter;
  * This tenant prefix is prepended to the kafka resources name in order to present an isolated
  * environment for each tenant.
  * <br/>
- * TODO prefix other resources e.g. group names, transaction ids
  * TODO disallow the use of topic uids belonging to one tenant by another.
  */
 public class MultiTenantTransformationFilter
@@ -132,7 +134,8 @@ public class MultiTenantTransformationFilter
         AddPartitionsToTxnRequestFilter, AddPartitionsToTxnResponseFilter,
         AddOffsetsToTxnRequestFilter,
         TxnOffsetCommitRequestFilter, TxnOffsetCommitResponseFilter {
-    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
+    private final String prefixResourceNameSeparator;
+    private String kafkaResourcePrefix;
 
     @Override
     public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
@@ -176,7 +179,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<ResponseFilterResult> onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response,
                                                                     FilterContext context) {
-        String tenantPrefix = getTenantPrefix(context);
+        String tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         response.topics().removeIf(topic -> !topic.name().startsWith(tenantPrefix)); // TODO: allow kafka internal topics to be returned?
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         return context.forwardResponse(header, response);
@@ -311,7 +314,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<ResponseFilterResult> onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response,
                                                                       FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(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);
@@ -321,7 +324,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request,
                                                                    FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setGroupId(tenantPrefix + request.groupId());
         return context.forwardRequest(header, request);
     }
@@ -329,7 +332,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request,
                                                                    FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setGroupId(tenantPrefix + request.groupId());
         return context.forwardRequest(header, request);
     }
@@ -337,7 +340,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request,
                                                                     FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setGroupId(tenantPrefix + request.groupId());
         return context.forwardRequest(header, request);
     }
@@ -345,7 +348,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request,
                                                                    FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setGroupId(tenantPrefix + request.groupId());
         return context.forwardRequest(header, request);
     }
@@ -403,7 +406,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onAddOffsetsToTxnRequest(short apiVersion, RequestHeaderData header, AddOffsetsToTxnRequestData request,
                                                                          FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         request.setGroupId(tenantPrefix + request.groupId());
         return context.forwardRequest(header, request);
@@ -412,7 +415,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<RequestFilterResult> onTxnOffsetCommitRequest(short apiVersion, RequestHeaderData header, TxnOffsetCommitRequestData request,
                                                                          FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         request.setGroupId(tenantPrefix + request.groupId());
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
@@ -429,7 +432,7 @@ public class MultiTenantTransformationFilter
     @Override
     public CompletionStage<ResponseFilterResult> onListTransactionsResponse(short apiVersion, ResponseHeaderData header, ListTransactionsResponseData response,
                                                                             FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         var filteredTransactions = response.transactionStates().stream().filter(listedTxn -> listedTxn.transactionalId().startsWith(tenantPrefix)).toList();
         filteredTransactions.forEach(listedTxn -> removeTenantPrefix(context, listedTxn::transactionalId, listedTxn::setTransactionalId, false));
         response.setTransactionStates(filteredTransactions);
@@ -455,7 +458,7 @@ public class MultiTenantTransformationFilter
 
     @Override
     public CompletionStage<RequestFilterResult> onEndTxnRequest(short apiVersion, RequestHeaderData header, EndTxnRequestData request, FilterContext context) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         return context.forwardRequest(header, request);
     }
@@ -469,7 +472,7 @@ public class MultiTenantTransformationFilter
     }
 
     private String applyTenantPrefix(FilterContext context, String clientSideName) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         return tenantPrefix + clientSideName;
     }
 
@@ -483,20 +486,32 @@ public class MultiTenantTransformationFilter
     }
 
     private String removeTenantPrefix(FilterContext context, String brokerSideName) {
-        var tenantPrefix = getTenantPrefix(context);
+        var tenantPrefix = createKafkaResourcePrefixIfNecessary(context);
         return brokerSideName.substring(tenantPrefix.length());
     }
 
-    private static String getTenantPrefix(FilterContext context) {
-        // TODO naive - POC implementation uses virtual cluster name as a tenant prefix
-        var virtualClusterName = context.getVirtualClusterName();
-        if (virtualClusterName == null) {
-            throw new IllegalStateException("This filter requires that the virtual cluster has a name");
+    private String createKafkaResourcePrefixIfNecessary(FilterContext context) {
+        if (kafkaResourcePrefix == null) {
+            // TODO naive - POC implementation uses virtual cluster name as a tenant prefix
+            var virtualClusterName = context.getVirtualClusterName();
+            if (virtualClusterName == null) {
+                throw new IllegalStateException("This filter requires that the virtual cluster has a name");
+            }
+            kafkaResourcePrefix = virtualClusterName + prefixResourceNameSeparator;
+
+            // note: Kafka validates consumer group names using this method too
+            Topic.validate(kafkaResourcePrefix,
+                    "Kafka resource prefix for virtual cluster '%s'".formatted(virtualClusterName),
+                    message -> {
+                        throw new IllegalStateException(message);
+                    });
         }
-        return virtualClusterName + "-";
+        return kafkaResourcePrefix;
     }
 
-    public MultiTenantTransformationFilter() {
+    public MultiTenantTransformationFilter(@NonNull MultiTenantConfig configuration) {
+        Objects.requireNonNull(configuration);
+        this.prefixResourceNameSeparator = configuration.prefixResourceNameSeparator();
     }
 
 }
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactory.java b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactory.java
index 56563837..244eabe5 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactory.java
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactory.java
@@ -6,20 +6,25 @@
 
 package io.kroxylicious.proxy.filter.multitenant;
 
+import java.util.Objects;
+
 import io.kroxylicious.proxy.filter.FilterFactory;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
+import io.kroxylicious.proxy.filter.multitenant.config.MultiTenantConfig;
 import io.kroxylicious.proxy.plugin.Plugin;
 
-@Plugin(configType = Void.class)
-public class MultiTenantTransformationFilterFactory implements FilterFactory<Void, Void> {
+@Plugin(configType = MultiTenantConfig.class)
+public class MultiTenantTransformationFilterFactory implements FilterFactory<MultiTenantConfig, MultiTenantConfig> {
+
+    private static final MultiTenantConfig DEFAULT_TENANT_CONFIG = new MultiTenantConfig(null);
 
     @Override
-    public Void initialize(FilterFactoryContext context, Void config) {
-        return null;
+    public MultiTenantConfig initialize(FilterFactoryContext context, MultiTenantConfig config) {
+        return config;
     }
 
     @Override
-    public MultiTenantTransformationFilter createFilter(FilterFactoryContext context, Void configuration) {
-        return new MultiTenantTransformationFilter();
+    public MultiTenantTransformationFilter createFilter(FilterFactoryContext context, MultiTenantConfig configuration) {
+        return new MultiTenantTransformationFilter(Objects.requireNonNullElse(configuration, DEFAULT_TENANT_CONFIG));
     }
 }
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/config/MultiTenantConfig.java b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/config/MultiTenantConfig.java
new file mode 100644
index 00000000..d70e2137
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/config/MultiTenantConfig.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.multitenant.config;
+
+import java.util.Objects;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for the Multi Tenant Filter.
+ * @param prefixResourceNameSeparator separator character used to form the prefixed resource name, if not null
+ * a default is used.
+ */
+public record MultiTenantConfig(String prefixResourceNameSeparator) {
+
+    public static final String DEFAULT_SEPARATOR = "-";
+
+    public MultiTenantConfig(@JsonProperty(required = false) String prefixResourceNameSeparator) {
+        this.prefixResourceNameSeparator = Objects.requireNonNullElse(prefixResourceNameSeparator, DEFAULT_SEPARATOR);
+    }
+}


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantFilterFactoryTest.java b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantFilterFactoryTest.java
index cd93dae2..fc46a17a 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantFilterFactoryTest.java
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantFilterFactoryTest.java
@@ -11,6 +11,7 @@ import org.mockito.Mockito;
 
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
+import io.kroxylicious.proxy.filter.multitenant.config.MultiTenantConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -19,8 +20,8 @@ class MultiTenantFilterFactoryTest {
     @Test
     void testGetInstance() {
         MultiTenantTransformationFilterFactory factory = new MultiTenantTransformationFilterFactory();
-        Filter filter = factory.createFilter(Mockito.mock(FilterFactoryContext.class), null);
+        Filter filter = factory.createFilter(Mockito.mock(FilterFactoryContext.class), Mockito.mock(MultiTenantConfig.class));
         assertThat(filter).isNotNull().isInstanceOf(MultiTenantTransformationFilter.class);
     }
 
-}
\ No newline at end of file
+}
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactoryFilterTest.java b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactoryFilterTest.java
index 9a8ff229..dbc43e1d 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactoryFilterTest.java
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterFactoryFilterTest.java
@@ -39,6 +39,7 @@ import io.kroxylicious.proxy.filter.FilterFactoryContext;
 import io.kroxylicious.proxy.filter.FilterInvoker;
 import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.filter.ResponseFilterResult;
+import io.kroxylicious.proxy.filter.multitenant.config.MultiTenantConfig;
 import io.kroxylicious.test.requestresponsetestdef.ApiMessageTestDef;
 import io.kroxylicious.test.requestresponsetestdef.RequestResponseTestDef;
 
@@ -69,7 +70,7 @@ class MultiTenantTransformationFilterFactoryFilterTest {
         return resources;
     }
 
-    private final MultiTenantTransformationFilter filter = new MultiTenantTransformationFilter();
+    private final MultiTenantTransformationFilter filter = new MultiTenantTransformationFilter(new MultiTenantConfig(null));
 
     private final FilterInvoker invoker = getOnlyElement(FilterAndInvoker.build(filter)).invoker();
 
@@ -153,6 +154,7 @@ class MultiTenantTransformationFilterFactoryFilterTest {
     @Test
     void testContributor() {
         MultiTenantTransformationFilterFactory factory = new MultiTenantTransformationFilterFactory();
-        assertThat(factory.createFilter(Mockito.mock(FilterFactoryContext.class), null)).isInstanceOf(MultiTenantTransformationFilter.class);
+        assertThat(factory.createFilter(Mockito.mock(FilterFactoryContext.class), Mockito.mock(MultiTenantConfig.class)))
+                .isInstanceOf(MultiTenantTransformationFilter.class);
     }
 }
diff --git a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
index b9b3b196..4231fc04 100644
--- a/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
+++ b/kroxylicious-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
@@ -6,20 +6,29 @@
 
 package io.kroxylicious.proxy.filter.multitenant;
 
+import java.util.Objects;
+
+import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.multitenant.config.MultiTenantConfig;
+import io.kroxylicious.test.condition.kafka.FetchResponseDataCondition;
 
 import static io.kroxylicious.test.condition.kafka.ApiMessageCondition.forApiKey;
 import static io.kroxylicious.test.condition.kafka.ProduceRequestDataCondition.produceRequestMatching;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.assertArg;
 import static org.mockito.Mockito.verify;
@@ -28,23 +37,17 @@ import static org.mockito.Mockito.when;
 @ExtendWith(MockitoExtension.class)
 class MultiTenantTransformationFilterTest {
 
-    private MultiTenantTransformationFilter multiTenantTransformationFilter;
+    private static final String VIRTUAL_CLUSTER_NAME = "vc1";
+    private static final String TEST_TOPIC = "testTopic";
     @Mock
     private FilterContext filterContext;
 
-    @BeforeEach
-    void setUp() {
-        multiTenantTransformationFilter = new MultiTenantTransformationFilter();
-    }
-
     @Test
     void shouldReWriteTopic() {
+        var multiTenantTransformationFilter = new MultiTenantTransformationFilter(new MultiTenantConfig(null));
         // Given
-        when(filterContext.getVirtualClusterName()).thenReturn("vc1");
-        final ProduceRequestData request = new ProduceRequestData();
-        final ProduceRequestData.TopicProduceData topicData = new ProduceRequestData.TopicProduceData();
-        topicData.setName("testTopic");
-        request.topicData().add(topicData);
+        when(filterContext.getVirtualClusterName()).thenReturn(VIRTUAL_CLUSTER_NAME);
+        var request = createProduceRequest(TEST_TOPIC);
         // When
         multiTenantTransformationFilter.onProduceRequest(
                 ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), request, filterContext);
@@ -55,8 +58,96 @@ class MultiTenantTransformationFilterTest {
                         .is(forApiKey(ApiKeys.PRODUCE))
                         .is(produceRequestMatching(produceRequestData -> produceRequestData.topicData()
                                 .stream()
-                                .allMatch(topicProduceData -> topicProduceData.name().equals("vc1-testTopic")))
-                        )));
+                                .allMatch(topicProduceData -> topicProduceData.name().equals("vc1-testTopic"))))));
 
     }
+
+    @ParameterizedTest
+    @ValueSource(strings = { ".", "_", "", "-.-" })
+    @NullSource
+    void produceWithCustomSeparator(String separator) {
+        var expectedServerTopicName = "%s%s%s".formatted(VIRTUAL_CLUSTER_NAME, Objects.requireNonNullElse(separator, MultiTenantConfig.DEFAULT_SEPARATOR), TEST_TOPIC);
+        var multiTenantTransformationFilter = new MultiTenantTransformationFilter(new MultiTenantConfig(separator));
+        // Given
+        when(filterContext.getVirtualClusterName()).thenReturn(VIRTUAL_CLUSTER_NAME);
+        var request = createProduceRequest(TEST_TOPIC);
+        // When
+        multiTenantTransformationFilter.onProduceRequest(
+                ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), request, filterContext);
+
+        // Then
+        verify(filterContext).forwardRequest(any(RequestHeaderData.class), assertArg(
+                apiMessage -> assertThat(apiMessage)
+                        .is(forApiKey(ApiKeys.PRODUCE))
+                        .is(produceRequestMatching(produceRequestData -> produceRequestData.topicData()
+                                .stream()
+                                .allMatch(topicProduceData -> topicProduceData.name().equals(expectedServerTopicName))))));
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = { ".", "_", "", "-.-" })
+    @NullSource
+    void fetchWithCustomSeparator(String separator) {
+        var serverTopic = "%s%s%s".formatted(VIRTUAL_CLUSTER_NAME, Objects.requireNonNullElse(separator, MultiTenantConfig.DEFAULT_SEPARATOR), TEST_TOPIC);
+        var multiTenantTransformationFilter = new MultiTenantTransformationFilter(new MultiTenantConfig(separator));
+        // Given
+        when(filterContext.getVirtualClusterName()).thenReturn(VIRTUAL_CLUSTER_NAME);
+        var response = createFetchResponseData(serverTopic);
+        // When
+        multiTenantTransformationFilter.onFetchResponse(
+                FetchResponseData.HIGHEST_SUPPORTED_VERSION, new ResponseHeaderData(), response, filterContext);
+
+        // Then
+        verify(filterContext).forwardResponse(any(ResponseHeaderData.class), assertArg(
+                apiMessage -> assertThat(apiMessage)
+                        .is(forApiKey(ApiKeys.FETCH))
+                        .is(FetchResponseDataCondition.fetchResponseMatching(fetchResponseData -> fetchResponseData.responses()
+                                .stream()
+                                .allMatch(topicFetchData -> topicFetchData.topic().equals(TEST_TOPIC))))));
+    }
+
+    @Test
+    void illegalKafkaResourceCharsInVirtualClusterNameDetected() {
+        var multiTenantTransformationFilter = new MultiTenantTransformationFilter(new MultiTenantConfig(MultiTenantConfig.DEFAULT_SEPARATOR));
+        // Given
+        when(filterContext.getVirtualClusterName()).thenReturn("$badvcn$");
+        var request = createProduceRequest(TEST_TOPIC);
+        var header = new RequestHeaderData();
+        // When
+        assertThatThrownBy(() -> {
+            multiTenantTransformationFilter.onProduceRequest(
+                    ProduceRequestData.HIGHEST_SUPPORTED_VERSION, header, request, filterContext);
+        }).isInstanceOf(IllegalStateException.class);
+    }
+
+    @Test
+    void illegalKafkaResourceCharsInSeparatorDetected() {
+        var multiTenantTransformationFilter = new MultiTenantTransformationFilter(new MultiTenantConfig("$bad$"));
+        // Given
+        when(filterContext.getVirtualClusterName()).thenReturn(VIRTUAL_CLUSTER_NAME);
+        var request = createProduceRequest(TEST_TOPIC);
+        var header = new RequestHeaderData();
+        // When
+        assertThatThrownBy(() -> {
+            multiTenantTransformationFilter.onProduceRequest(
+                    ProduceRequestData.HIGHEST_SUPPORTED_VERSION, header, request, filterContext);
+        }).isInstanceOf(IllegalStateException.class);
+    }
+
+    private ProduceRequestData createProduceRequest(String topic) {
+        final ProduceRequestData request = new ProduceRequestData();
+        final ProduceRequestData.TopicProduceData topicData = new ProduceRequestData.TopicProduceData();
+        topicData.setName(topic);
+        request.topicData().add(topicData);
+        return request;
+    }
+
+    private FetchResponseData createFetchResponseData(String topic) {
+        var response = new FetchResponseData();
+        var topicResponse = new FetchResponseData.FetchableTopicResponse();
+        topicResponse.setTopic(topic);
+        response.responses().add(topicResponse);
+        return response;
+    }
+
 }
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
index 1fcabbaf..03e9a31b 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
@@ -90,6 +90,12 @@ public abstract class BaseMultiTenantIT extends BaseIT {
     }
 
     static ConfigurationBuilder getConfig(KafkaCluster cluster, KeytoolCertificateGenerator certificateGenerator) {
+        return getConfig(cluster, certificateGenerator, null);
+    }
+
+    static ConfigurationBuilder getConfig(KafkaCluster cluster, KeytoolCertificateGenerator certificateGenerator, Map<String, Object> filterConfig) {
+        var filterBuilder = new FilterDefinitionBuilder(MultiTenantTransformationFilterFactory.class.getName());
+        Optional.ofNullable(filterConfig).ifPresent(filterBuilder::withConfig);
         return new ConfigurationBuilder()
                 .addToVirtualClusters(TENANT_1_CLUSTER, new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -121,7 +127,7 @@ public abstract class BaseMultiTenantIT extends BaseIT {
                         .endKeyStoreKey()
                         .endTls()
                         .build())
-                .addToFilters(new FilterDefinitionBuilder(MultiTenantTransformationFilterFactory.class.getName()).build());
+                .addToFilters(filterBuilder.build());
     }
 
     Consumer<String, String> getConsumerWithConfig(KroxyliciousTester tester, String virtualCluster, String groupId, Map<String, Object> baseConfig,
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index ca9e5897..41f4dc4b 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -24,6 +24,7 @@ import java.util.stream.Stream;
 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.KafkaAdminClient;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.admin.TopicListing;
@@ -46,8 +47,6 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.config.ConfigurationBuilder;
 import io.kroxylicious.test.tester.KroxyliciousTester;
@@ -65,8 +64,6 @@ import static org.awaitility.Awaitility.await;
 @ExtendWith(KafkaClusterExtension.class)
 class MultiTenantIT extends BaseMultiTenantIT {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantIT.class);
-
     private static final String TOPIC_1 = "my-test-topic";
     private static final NewTopic NEW_TOPIC_1 = new NewTopic(TOPIC_1, 1, (short) 1);
     private static final String TOPIC_2 = "other-test-topic";
@@ -364,6 +361,24 @@ class MultiTenantIT extends BaseMultiTenantIT {
         }
     }
 
+    @Test
+    void tenantPrefixUsingCustomSeparator(KafkaCluster cluster, KafkaAdminClient directAdmin) {
+        var separator = ".";
+        var config = getConfig(cluster, this.certificateGenerator, Map.of("prefixResourceNameSeparator", separator));
+        try (var tester = kroxyliciousTester(config);
+                var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
+            createTopics(admin, NEW_TOPIC_1);
+        }
+
+        var expectedTargetClusterTopicName = "%s%s%s".formatted(TENANT_1_CLUSTER, separator, TOPIC_1);
+        await().atMost(Duration.ofSeconds(5)).ignoreExceptions().untilAsserted(() -> {
+            var listTopicsResult = directAdmin.listTopics();
+            var names = listTopicsResult.names().get();
+            assertThat(names).contains(expectedTargetClusterTopicName);
+        });
+
+    }
+
     // ========================================================
     // UTILS
     // ========================================================