Repository: kroxylicious/kroxylicious
Pull request #210 (2023-03-20 22:39:49)

Generate code for the JSON/Kafka request/response converters
### Type of change

_Select the type of your PR_

- Refactoring

### Description

Generating the JSON/Kafka request/response converters used by the multi-tenant unit tests by hand was already getting tedious.

### Additional Context

_Why are you making this pull request?_



Patch:
diff --git a/kroxylicious-multitenant/pom.xml b/kroxylicious-multitenant/pom.xml
index 7796be7a..03b735ff 100644
--- a/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-multitenant/pom.xml
@@ -46,6 +46,11 @@
             <artifactId>jackson-dataformat-yaml</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-slf4j-impl</artifactId>
@@ -77,4 +82,75 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>unpack-message-specs</id>
+                        <goals>
+                            <goal>unpack</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <artifact>org.apache.kafka:kafka-clients:${kafka.version}</artifact>
+                            <includes>common/message/*.json</includes>
+                            <outputDirectory>${project.build.directory}/message-specs</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-krpc-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>generate-converters</id>
+                        <goals>
+                            <goal>generate-multi</goal>
+                        </goals>
+                        <phase>generate-test-sources</phase>
+                        <configuration>
+                            <messageSpecDirectory>${project.build.directory}/message-specs/common/message</messageSpecDirectory>
+                            <messageSpecFilter>*{Request,Response}.json</messageSpecFilter>
+                            <templateDirectory>${project.basedir}/src/test/templates</templateDirectory>
+                            <templateNames>KafkaApiMessageConverter.ftl</templateNames>
+                            <outputFilePattern>${templateName}.java</outputFilePattern>
+                            <outputPackage>io.kroxylicious.proxy.filter.multitenant</outputPackage>
+                            <outputDirectory>${project.build.directory}/generated-test-sources</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>add-source</id>
+                        <phase>generate-test-sources</phase>
+                        <goals>
+                            <goal>add-test-source</goal>
+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>${project.build.directory}/generated-test-sources</source>
+                            </sources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-javadoc-plugin</artifactId>
+                <configuration>
+                    <!-- Javadoc generation fails to resolve jackson databind.  KW doesn't understand why -->
+                    <!-- Turn off javadoc generation for this module to work-around -->
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>


Test patch:
diff --git a/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java b/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
index f8c8fd11..13c2597a 100644
--- a/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
+++ b/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
@@ -10,7 +10,6 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.file.Path;
 import java.util.Arrays;
-import java.util.Map;
 import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.function.Predicate;
@@ -18,42 +17,6 @@ import java.util.regex.Pattern;
 import java.util.stream.Stream;
 
 import org.apache.kafka.common.message.ApiMessageType;
-import org.apache.kafka.common.message.CreateTopicsRequestData;
-import org.apache.kafka.common.message.CreateTopicsRequestDataJsonConverter;
-import org.apache.kafka.common.message.CreateTopicsResponseData;
-import org.apache.kafka.common.message.CreateTopicsResponseDataJsonConverter;
-import org.apache.kafka.common.message.DeleteTopicsRequestData;
-import org.apache.kafka.common.message.DeleteTopicsRequestDataJsonConverter;
-import org.apache.kafka.common.message.DeleteTopicsResponseData;
-import org.apache.kafka.common.message.DeleteTopicsResponseDataJsonConverter;
-import org.apache.kafka.common.message.FetchRequestData;
-import org.apache.kafka.common.message.FetchRequestDataJsonConverter;
-import org.apache.kafka.common.message.FetchResponseData;
-import org.apache.kafka.common.message.FetchResponseDataJsonConverter;
-import org.apache.kafka.common.message.ListOffsetsRequestData;
-import org.apache.kafka.common.message.ListOffsetsRequestDataJsonConverter;
-import org.apache.kafka.common.message.ListOffsetsResponseData;
-import org.apache.kafka.common.message.ListOffsetsResponseDataJsonConverter;
-import org.apache.kafka.common.message.MetadataRequestData;
-import org.apache.kafka.common.message.MetadataRequestDataJsonConverter;
-import org.apache.kafka.common.message.MetadataResponseData;
-import org.apache.kafka.common.message.MetadataResponseDataJsonConverter;
-import org.apache.kafka.common.message.OffsetCommitRequestData;
-import org.apache.kafka.common.message.OffsetCommitRequestDataJsonConverter;
-import org.apache.kafka.common.message.OffsetCommitResponseData;
-import org.apache.kafka.common.message.OffsetCommitResponseDataJsonConverter;
-import org.apache.kafka.common.message.OffsetFetchRequestData;
-import org.apache.kafka.common.message.OffsetFetchRequestDataJsonConverter;
-import org.apache.kafka.common.message.OffsetFetchResponseData;
-import org.apache.kafka.common.message.OffsetFetchResponseDataJsonConverter;
-import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
-import org.apache.kafka.common.message.OffsetForLeaderEpochRequestDataJsonConverter;
-import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
-import org.apache.kafka.common.message.OffsetForLeaderEpochResponseDataJsonConverter;
-import org.apache.kafka.common.message.ProduceRequestData;
-import org.apache.kafka.common.message.ProduceRequestDataJsonConverter;
-import org.apache.kafka.common.message.ProduceResponseData;
-import org.apache.kafka.common.message.ProduceResponseDataJsonConverter;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -76,6 +39,8 @@ import io.kroxylicious.proxy.filter.KrpcFilterContext;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
+import static io.kroxylicious.proxy.filter.multitenant.KafkaApiMessageConverter.requestConverterFor;
+import static io.kroxylicious.proxy.filter.multitenant.KafkaApiMessageConverter.responseConverterFor;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -92,59 +57,6 @@ class MultiTenantTransformationFilterTest {
     private record RequestResponseTestDef(String testName, ApiMessageType apiKey, RequestHeaderData header, ApiMessageTestDef request, ApiMessageTestDef response) {
     }
 
-    private record Converters(BiFunction<JsonNode, Short, ApiMessage> requestReader,
-                              BiFunction<JsonNode, Short, ApiMessage> responseReader,
-                              BiFunction<ApiMessage, Short, JsonNode> requestWriter,
-                              BiFunction<ApiMessage, Short, JsonNode> responseWriters) {
-    }
-
-    private static final Map<ApiMessageType, Converters> converters = Map.of(
-            ApiMessageType.CREATE_TOPICS, new Converters(
-                    CreateTopicsRequestDataJsonConverter::read,
-                    CreateTopicsResponseDataJsonConverter::read,
-                    (o, ver) -> CreateTopicsRequestDataJsonConverter.write(((CreateTopicsRequestData) o), ver),
-                    (o, ver) -> CreateTopicsResponseDataJsonConverter.write(((CreateTopicsResponseData) o), ver)),
-            ApiMessageType.DELETE_TOPICS, new Converters(
-                    DeleteTopicsRequestDataJsonConverter::read,
-                    DeleteTopicsResponseDataJsonConverter::read,
-                    (o, ver) -> DeleteTopicsRequestDataJsonConverter.write(((DeleteTopicsRequestData) o), ver),
-                    (o, ver) -> DeleteTopicsResponseDataJsonConverter.write(((DeleteTopicsResponseData) o), ver)),
-            ApiMessageType.METADATA, new Converters(
-                    MetadataRequestDataJsonConverter::read,
-                    MetadataResponseDataJsonConverter::read,
-                    (o, ver) -> MetadataRequestDataJsonConverter.write(((MetadataRequestData) o), ver),
-                    (o, ver) -> MetadataResponseDataJsonConverter.write(((MetadataResponseData) o), ver)),
-            ApiMessageType.PRODUCE, new Converters(
-                    ProduceRequestDataJsonConverter::read,
-                    ProduceResponseDataJsonConverter::read,
-                    (o, ver) -> ProduceRequestDataJsonConverter.write(((ProduceRequestData) o), ver),
-                    (o, ver) -> ProduceResponseDataJsonConverter.write(((ProduceResponseData) o), ver)),
-            ApiMessageType.LIST_OFFSETS, new Converters(
-                    ListOffsetsRequestDataJsonConverter::read,
-                    ListOffsetsResponseDataJsonConverter::read,
-                    (o, ver) -> ListOffsetsRequestDataJsonConverter.write(((ListOffsetsRequestData) o), ver),
-                    (o, ver) -> ListOffsetsResponseDataJsonConverter.write(((ListOffsetsResponseData) o), ver)),
-            ApiMessageType.OFFSET_FETCH, new Converters(
-                    OffsetFetchRequestDataJsonConverter::read,
-                    OffsetFetchResponseDataJsonConverter::read,
-                    (o, ver) -> OffsetFetchRequestDataJsonConverter.write(((OffsetFetchRequestData) o), ver),
-                    (o, ver) -> OffsetFetchResponseDataJsonConverter.write(((OffsetFetchResponseData) o), ver)),
-            ApiMessageType.OFFSET_FOR_LEADER_EPOCH, new Converters(
-                    OffsetForLeaderEpochRequestDataJsonConverter::read,
-                    OffsetForLeaderEpochResponseDataJsonConverter::read,
-                    (o, ver) -> OffsetForLeaderEpochRequestDataJsonConverter.write(((OffsetForLeaderEpochRequestData) o), ver),
-                    (o, ver) -> OffsetForLeaderEpochResponseDataJsonConverter.write(((OffsetForLeaderEpochResponseData) o), ver)),
-            ApiMessageType.OFFSET_COMMIT, new Converters(
-                    OffsetCommitRequestDataJsonConverter::read,
-                    OffsetCommitResponseDataJsonConverter::read,
-                    (o, ver) -> OffsetCommitRequestDataJsonConverter.write(((OffsetCommitRequestData) o), ver),
-                    (o, ver) -> OffsetCommitResponseDataJsonConverter.write(((OffsetCommitResponseData) o), ver)),
-            ApiMessageType.FETCH, new Converters(
-                    FetchRequestDataJsonConverter::read,
-                    FetchResponseDataJsonConverter::read,
-                    (o, ver) -> FetchRequestDataJsonConverter.write(((FetchRequestData) o), ver),
-                    (o, ver) -> FetchResponseDataJsonConverter.write(((FetchResponseData) o), ver)));
-
     private static Stream<RequestResponseTestDef> requestResponseTestDefinitions() throws Exception {
         var resourceInfoStream = ClassPath.from(MultiTenantTransformationFilterTest.class.getClassLoader()).getResources().stream()
                 .filter(ri -> TEST_RESOURCE_FILTER.matcher(ri.getResourceName()).matches()).toList();
@@ -191,9 +103,10 @@ class MultiTenantTransformationFilterTest {
 
         var testName = String.format("%s,%s,v%d[%s]", source, messageType, versionNode.intValue(),
                 Optional.ofNullable(descriptionNode).map(JsonNode::asText).orElse("-"));
-        var conv = converters.get(messageType);
-        var request = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("request"), conv.requestReader());
-        var response = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("response"), conv.responseReader());
+        var requestConverter = requestConverterFor(messageType);
+        var responseConverter = responseConverterFor(messageType);
+        var request = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("request"), requestConverter.reader());
+        var response = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("response"), responseConverter.reader());
         return new RequestResponseTestDef(testName, messageType, header, request, response);
     }
 
@@ -226,7 +139,7 @@ class MultiTenantTransformationFilterTest {
     void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef) {
         var request = requestTestDef.message();
         // marshalled the request object back to json, this is used for the comparison later.
-        var requestWriter = converters.get(apiMessageType).requestWriter();
+        var requestWriter = requestConverterFor(apiMessageType).writer();
         var marshalled = requestWriter.apply(request, header.requestApiVersion());
 
         filter.onRequest(ApiKeys.forId(apiMessageType.apiKey()), header, request, context);
@@ -245,7 +158,7 @@ class MultiTenantTransformationFilterTest {
     void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
         var response = responseTestDef.message();
         // marshalled the response object back to json, this is used for comparison later.
-        var responseWriter = converters.get(apiMessageType).responseWriters();
+        var responseWriter = responseConverterFor(apiMessageType).writer();
 
         var marshalled = responseWriter.apply(response, header.requestApiVersion());
 
diff --git a/kroxylicious-multitenant/src/test/templates/KafkaApiMessageConverter.ftl b/kroxylicious-multitenant/src/test/templates/KafkaApiMessageConverter.ftl
new file mode 100644
index 00000000..bece75d8
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/templates/KafkaApiMessageConverter.ftl
@@ -0,0 +1,71 @@
+<#--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+
+package ${outputPackage};
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+<#list messageSpecs as messageSpec>
+import org.apache.kafka.common.message.${messageSpec.name}Data;
+import org.apache.kafka.common.message.${messageSpec.name}DataJsonConverter;
+</#list>
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+
+
+public class KafkaApiMessageConverter {
+
+    public record Converter(BiFunction<JsonNode, Short, ApiMessage> reader,
+                            BiFunction<ApiMessage, Short, JsonNode> writer) {
+    }
+
+    private static final Map<ApiMessageType, Converter> requestConverters;
+    private static final Map<ApiMessageType, Converter> responseConverters;
+
+    static {
+        var reqc = new HashMap<ApiMessageType, Converter>();
+        var resc = new HashMap<ApiMessageType, Converter>();
+
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'request'>
+        reqc.put(ApiMessageType.${retrieveApiKey(messageSpec)}, new Converter(
+                    ${messageSpec.name}DataJsonConverter::read,
+                (o, ver) -> ${messageSpec.name}DataJsonConverter.write(((${messageSpec.name}Data) o), ver)));
+    </#if>
+    <#if messageSpec.type?lower_case == 'response'>
+        resc.put(ApiMessageType.${retrieveApiKey(messageSpec)}, new Converter(
+                    ${messageSpec.name}DataJsonConverter::read,
+                (o, ver) -> ${messageSpec.name}DataJsonConverter.write(((${messageSpec.name}Data) o), ver)));
+    </#if>
+</#list>
+        requestConverters = Collections.unmodifiableMap(reqc);
+        responseConverters = Collections.unmodifiableMap(resc);
+    }
+
+    public static Converter requestConverterFor(ApiMessageType apiMessageType) {
+        var converter = requestConverters.get(apiMessageType);
+        if (converter == null) {
+            throw new IllegalArgumentException("no request converter registered for " + apiMessageType);
+        }
+        return converter;
+    }
+
+    public static Converter responseConverterFor(ApiMessageType apiMessageType) {
+        var converter = responseConverters.get(apiMessageType);
+        if (converter == null) {
+            throw new IllegalArgumentException("no response converter registered for " + apiMessageType);
+        }
+        return converter;
+    }
+}