Repository: kroxylicious/kroxylicious
Pull request #144 (2023-02-19 22:13:13)

Change kroxylicious Future to jdk CompletionStage in KrpcFilterContext
Why:
We are working to extract an filter api that Filter Authers will use to build custom filters. This currently would include the kroxylicious futures classes, so adds the extra complexity of extracting that into a module or making kroxy futures part of the filter-api. With this change we use a JDK supplied CompletionStage in the KrpcFilterContext


Patch:
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index a1bcb9af..20b33552 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -5,12 +5,12 @@
  */
 package io.kroxylicious.proxy.filter;
 
+import java.util.concurrent.CompletionStage;
+
 import org.apache.kafka.common.protocol.ApiMessage;
 
 import io.netty.buffer.ByteBuf;
 
-import io.kroxylicious.proxy.future.Future;
-
 /**
  * A context to allow filters to interact with other filters and the pipeline.
  */
@@ -51,7 +51,7 @@ public interface KrpcFilterContext {
      * @param request The request to send.
      * @param <T> The type of the response
      */
-    <T extends ApiMessage> Future<T> sendRequest(short apiVersion, ApiMessage request);
+    <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request);
 
     /**
      * Send a response towards the client, invoking downstream filters.
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
index 6c8bb310..e82dce65 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.internal;
 
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.common.errors.TimeoutException;
@@ -22,7 +23,6 @@ import io.netty.channel.ChannelPromise;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.frame.DecodedFrame;
-import io.kroxylicious.proxy.future.Future;
 import io.kroxylicious.proxy.future.Promise;
 
 /**
@@ -106,7 +106,7 @@ class DefaultFilterContext implements KrpcFilterContext {
     }
 
     @Override
-    public <T extends ApiMessage> Future<T> sendRequest(short apiVersion, ApiMessage message) {
+    public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage message) {
         short key = message.apiKey();
         var apiKey = ApiKeys.forId(key);
         short headerVersion = apiKey.requestHeaderVersion(apiVersion);
@@ -154,7 +154,7 @@ class DefaultFilterContext implements KrpcFilterContext {
             LOGGER.debug("{}: Timing out {} request after {}ms", channelContext, apiKey, timeoutMs);
             filterPromise.tryFail(new TimeoutException());
         }, timeoutMs, TimeUnit.MILLISECONDS);
-        return filterPromise.future();
+        return filterPromise.future().toCompletionStage();
     }
 
     /**
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
index 1baccba5..dd00b2a3 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
@@ -29,6 +29,7 @@ import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.future.Future;
 import io.kroxylicious.proxy.internal.util.NettyMemoryRecords;
 
 /**
@@ -81,9 +82,9 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
         if (!requestTopics.isEmpty()) {
             LOGGER.debug("Fetch response contains {} unknown topic ids, lookup via Metadata request: {}", requestTopics.size(), requestTopics);
             // TODO Can't necessarily use HIGHEST_SUPPORTED_VERSION, must use highest supported version
-            context.<MetadataResponseData> sendRequest(MetadataRequestData.HIGHEST_SUPPORTED_VERSION,
+            Future.fromCompletionStage(context.<MetadataResponseData> sendRequest(MetadataRequestData.HIGHEST_SUPPORTED_VERSION,
                     new MetadataRequestData()
-                            .setTopics(requestTopics))
+                            .setTopics(requestTopics)))
                     .map(metadataResponse -> {
                         Map<Uuid, String> uidToName = metadataResponse.topics().stream().collect(Collectors.toMap(ti -> ti.topicId(), ti -> ti.name()));
                         LOGGER.debug("Metadata response yields {}, updating original Fetch response", uidToName);


Test patch:
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
index fc5a1c41..959134df 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -108,7 +108,7 @@ public class FilterHandlerTest extends FilterHarness {
         ApiVersionsRequestFilter filter = (request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
-            fut[0] = context.sendRequest((short) 3, body);
+            fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
         };
 
         buildChannel(filter);
@@ -143,7 +143,7 @@ public class FilterHandlerTest extends FilterHarness {
         ApiVersionsRequestFilter filter = (request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
-            fut[0] = context.sendRequest((short) 3, body);
+            fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
         };
 
         buildChannel(filter);
@@ -169,7 +169,7 @@ public class FilterHandlerTest extends FilterHarness {
         ApiVersionsRequestFilter filter = (request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
-            fut[0] = context.sendRequest((short) 3, body);
+            fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
         };
 
         buildChannel(filter, 50L);