name retriever short-circuit on empty names request

If a client requests a mapping for the empty set, immediately supply
them with an empty mapping, using the filter dispatch executor
associated with the filter (until now we have leant on sendRequest
to control which thread chained futures will execute on).

Why:
There is nothing to retrieve, so making a Metadata Request is a waste of
time and resources.

In some filters it may be convenient to always request topic names, to
simplify async logic, even if the topic name set is empty.

Signed-off-by: Robert Young <[email protected]>

Author: robobario

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java

@@ -110,6 +110,26 @@ class FilterIT {
     public static final String TOPIC_ID_LOOKUP_FILTER_NAME = "topicIdLookup";
     public static final String TOPIC_NAME_PREFIXER_FILTER_NAME = "topicNamePrefixer";
 
+    @Test
+    void filtersCanLookUpEmptyTopicNames(KafkaCluster cluster) {
+        NamedFilterDefinition namedFilterDefinition = new NamedFilterDefinitionBuilder(TOPIC_ID_LOOKUP_FILTER_NAME,
+                TopicIdToNameResponseStamper.class.getName())
+                .build();
+        var config = proxy(cluster)
+                .addToFilterDefinitions(namedFilterDefinition)
+                .addToDefaultFilters(namedFilterDefinition.name());
+
+        try (var tester = kroxyliciousTester(config);
+                var client = tester.simpleTestClient()) {
+            MetadataRequestData message = new MetadataRequestData();
+            message.unknownTaggedFields().add(
+                    new RawTaggedField(TopicIdToNameResponseStamper.TOPIC_ID_TAG, ("").getBytes(StandardCharsets.UTF_8)));
+            Response response = client.getSync(new Request(METADATA, METADATA.latestVersion(), "client", message));
+            // checking that the request/response flows through despite requesting an empty topic id list
+            assertThat(response).isNotNull();
+        }
+    }
+
     @Test
     void filtersCanLookUpTopicNames(KafkaCluster cluster, Topic topic1, Topic topic2) {
         Uuid topic1Id = topic1.topicId().orElseThrow();

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/testplugins/TopicIdToNameResponseStamper.java

@@ -17,6 +17,7 @@
 import java.util.stream.Stream;
 
 import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -26,6 +27,7 @@
 import io.kroxylicious.UnknownTaggedFields;
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.FilterDispatchExecutor;
 import io.kroxylicious.proxy.filter.FilterFactory;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
 import io.kroxylicious.proxy.filter.RequestFilter;
@@ -57,22 +59,31 @@ public Void initialize(FilterFactoryContext context, Void config) throws PluginC
 
     @Override
     public Filter createFilter(FilterFactoryContext context, Void initializationData) {
-        return new TopicIdToNameResponseStamperFilter();
+        return new TopicIdToNameResponseStamperFilter(context.filterDispatchExecutor());
     }
 
     static class TopicIdToNameResponseStamperFilter implements RequestFilter, ResponseFilter {
 
+        private final FilterDispatchExecutor filterDispatchExecutor;
         Map<Integer, TopicNameMapping> correlated = new HashMap<>();
 
+        TopicIdToNameResponseStamperFilter(FilterDispatchExecutor filterDispatchExecutor) {
+            this.filterDispatchExecutor = filterDispatchExecutor;
+        }
+
         @Override
         public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage request, FilterContext context) {
             List<String> list = UnknownTaggedFields.unknownTaggedFieldsToStrings(request, TOPIC_ID_TAG).toList();
             if (list.isEmpty()) {
-                return context.forwardRequest(header, request);
+                return context.requestFilterResultBuilder().errorResponse(header, request, new InvalidRequestException("no topic id tag")).withCloseConnection()
+                        .completed();
             }
 
-            Set<Uuid> uuids = Arrays.stream(list.getFirst().split(",")).map(Uuid::fromString).collect(Collectors.toSet());
+            Set<Uuid> uuids = Arrays.stream(list.getFirst().split(",")).filter(s -> !s.isEmpty()).map(Uuid::fromString).collect(Collectors.toSet());
             return context.topicNames(uuids).thenCompose(topicNames -> {
+                if (!filterDispatchExecutor.isInFilterDispatchThread()) {
+                    throw new IllegalStateException("work chained to topicNames future should execute in filter dispatch thread");
+                }
                 correlated.put(header.correlationId(), topicNames);
                 return context.forwardRequest(header, request);
             });

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java

@@ -622,7 +622,7 @@ public <M extends ApiMessage> CompletionStage<M> sendRequest(RequestHeaderData h
 
         @Override
         public CompletionStage<TopicNameMapping> topicNames(Collection<Uuid> topicIds) {
-            return new TopicNameRetriever(this).topicNames(topicIds);
+            return new TopicNameRetriever(this, Objects.requireNonNull(ctx).executor()).topicNames(topicIds);
         }
 
     }

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/TopicNameRetriever.java

@@ -11,8 +11,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
 
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.MetadataRequestData;
@@ -36,13 +38,18 @@ final class TopicNameRetriever {
     // Version 12 was the first version that uses topic ids.
     private static final short METADATA_API_VER_WITH_TOPIC_ID_SUPPORT = (short) 12;
     private final FilterContext filterContext;
+    private final Executor filterDispatchExecutor;
 
-    TopicNameRetriever(FilterContext filterContext) {
+    TopicNameRetriever(FilterContext filterContext, Executor filterDispatchExecutor) {
         this.filterContext = filterContext;
+        this.filterDispatchExecutor = filterDispatchExecutor;
     }
 
     CompletionStage<TopicNameMapping> topicNames(Collection<Uuid> topicIds) {
         Objects.requireNonNull(topicIds);
+        if (topicIds.isEmpty()) {
+            return CompletableFuture.supplyAsync(() -> MapTopicNameMapping.EMPTY, filterDispatchExecutor).minimalCompletionStage();
+        }
         CompletionStage<ApiMessage> apiMessageCompletionStage = requestTopicMetadata(topicIds);
         return apiMessageCompletionStage
                 .thenApply(message -> extractTopicNames(topicIds, message))
@@ -116,6 +123,8 @@ private static TopicNameMapping doExtractTopicNames(Collection<Uuid> topicIds, M
      */
     private record MapTopicNameMapping(Map<Uuid, String> topicNames, Map<Uuid, TopicNameMappingException> failures) implements TopicNameMapping {
 
+        private static final TopicNameMapping EMPTY = new MapTopicNameMapping(Map.of(), Map.of());
+
         private MapTopicNameMapping {
             Objects.requireNonNull(topicNames);
             Objects.requireNonNull(failures);

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/TopicNameRetrieverTest.java

@@ -16,6 +16,7 @@
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.protocol.Errors;
+import org.awaitility.core.InternalExecutorServiceFactory;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -33,6 +34,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
@@ -49,7 +52,7 @@ class TopicNameRetrieverTest {
 
     @BeforeEach
     void setUp() {
-        retriever = new TopicNameRetriever(filterContext);
+        retriever = new TopicNameRetriever(filterContext, InternalExecutorServiceFactory.sameThreadExecutorService());
     }
 
     @Test
@@ -68,6 +71,20 @@ void retrieveTopicNamesMapping() {
                 });
     }
 
+    @Test
+    void retrieveEmptyTopicNamesMapping() {
+        // when
+        CompletionStage<TopicNameMapping> topicNames = getTopicNamesMapping(Set.of());
+        // then
+        assertThat(topicNames.toCompletableFuture()).succeedsWithin(Duration.ZERO)
+                .satisfies(topicNamesMapping -> {
+                    assertThat(topicNamesMapping.anyFailures()).isFalse();
+                    assertThat(topicNamesMapping.topicNames()).isEmpty();
+                    assertThat(topicNamesMapping.failures()).isEmpty();
+                });
+        verify(filterContext, never()).sendRequest(any(), any());
+    }
+
     @Test
     void retrieveTopicNamesMappingServerPartialResponse() {
         // given