Refactor the handler for METADATA request

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ListPair.java

@@ -0,0 +1,62 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop;
+
+import io.streamnative.pulsar.handlers.kop.utils.CoreUtils;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * A wrapper of Map<Boolean, List<T>>, which can be returned by {@link java.util.stream.Collectors#groupingBy} with
+ * a {@link Function} that returns a Boolean.
+ *
+ * If we operates on the returned Map directly with {@link Map#get}, a null value might be returned. This class returns
+ * a empty map to avoid null check. In addition, the method name is more readable than `get(true)` or `get(false)`.
+ *
+ * @param <T>
+ */
+public class ListPair<T> {
+
+    private final Map<Boolean, List<T>> data;
+
+    public <R> ListPair<R> map(final Function<T, R> function) {
+        return of(CoreUtils.mapValue(data, list -> CoreUtils.listToList(list, function)));
+    }
+
+    public List<T> getSuccessfulList() {
+        return Optional.ofNullable(data.get(true)).orElse(Collections.emptyList());
+    }
+
+    public List<T> getFailedList() {
+        return Optional.ofNullable(data.get(false)).orElse(Collections.emptyList());
+    }
+
+    public static <T> ListPair<T> split(final Stream<T> stream,
+                                        final Function<T, Boolean> function) {
+        return ListPair.of(stream.collect(Collectors.groupingBy(function)));
+    }
+
+    public static <T> ListPair<T> of(final Map<Boolean, List<T>> data) {
+        return new ListPair<>(data);
+    }
+
+    private ListPair(final Map<Boolean, List<T>> data) {
+        this.data = (data != null) ? data : Collections.emptyMap();
+    }
+}

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java

@@ -0,0 +1,98 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop;
+
+import static org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata;
+import static org.apache.kafka.common.requests.MetadataResponse.TopicMetadata;
+
+import io.streamnative.pulsar.handlers.kop.utils.CoreUtils;
+import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.pulsar.common.naming.TopicName;
+
+/**
+ * The topic and its metadata (number of partitions or the error).
+ */
+@AllArgsConstructor
+@Getter
+public class TopicAndMetadata {
+
+    public static final int INVALID_PARTITIONS = -2;
+    public static final int AUTHORIZATION_FAILURE = -1;
+    public static final int NON_PARTITIONED_NUMBER = 0;
+
+    private final String topic;
+    private final int numPartitions;
+
+    public boolean isPartitionedTopic() {
+        return numPartitions > 0;
+    }
+
+    public boolean hasNoError() {
+        return numPartitions >= 0;
+    }
+
+    public Errors error() {
+        if (hasNoError()) {
+            return Errors.NONE;
+        } else if (numPartitions == AUTHORIZATION_FAILURE) {
+            return Errors.TOPIC_AUTHORIZATION_FAILED;
+        } else {
+            return Errors.UNKNOWN_TOPIC_OR_PARTITION;
+        }
+    }
+
+    public CompletableFuture<TopicMetadata> lookupAsync(
+            final Function<TopicName, CompletableFuture<PartitionMetadata>> lookupFunction,
+            final Function<String, String> getOriginalTopic,
+            final String metadataNamespace) {
+        return CoreUtils.waitForAll(stream()
+                .map(TopicName::get)
+                .map(lookupFunction)
+                .collect(Collectors.toList()), partitionMetadataList ->
+                new TopicMetadata(
+                        error(),
+                        getOriginalTopic.apply(topic),
+                        KopTopic.isInternalTopic(topic, metadataNamespace),
+                        partitionMetadataList
+                ));
+    }
+
+    public Stream<String> stream() {
+        if (numPartitions > 0) {
+            return IntStream.range(0, numPartitions)
+                    .mapToObj(i -> topic + "-partition-" + i);
+        } else {
+            return Stream.of(topic);
+        }
+    }
+
+    public TopicMetadata toTopicMetadata(final Function<String, String> getOriginalTopic,
+                                         final String metadataNamespace) {
+        return new TopicMetadata(
+                error(),
+                getOriginalTopic.apply(topic),
+                KopTopic.isInternalTopic(topic, metadataNamespace),
+                Collections.emptyList()
+        );
+    }
+}

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java

@@ -13,10 +13,14 @@
  */
 package io.streamnative.pulsar.handlers.kop.utils;
 
+import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -76,4 +80,28 @@ public static <K, V1, V2> Map<K, V2> mapKeyValue(Map<K, V1> map,
             ));
     }
 
+    public static <R, T> List<R> listToList(final List<T> list,
+                                            final Function<T, R> function) {
+        return list.stream().map(function).collect(Collectors.toList());
+    }
+
+    public static <K, V> Map<K, V> listToMap(final List<K> list,
+                                             final Function<K, V> function) {
+        return list.stream().collect(Collectors.toMap(key -> key, function));
+    }
+
+    public static <K, V, R> List<R> mapToList(final Map<K, V> map,
+                                              final BiFunction<K, V, R> function) {
+        return map.entrySet().stream().map(e -> function.apply(e.getKey(), e.getValue())).collect(Collectors.toList());
+    }
+
+    public static <T> CompletableFuture<Void> waitForAll(final Collection<CompletableFuture<T>> futures) {
+        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+    }
+
+    public static <T, R> CompletableFuture<R> waitForAll(final Collection<CompletableFuture<T>> futures,
+                                                         final Function<List<T>, R> function) {
+        return waitForAll(futures).thenApply(__ ->
+                function.apply(futures.stream().map(CompletableFuture::join).collect(Collectors.toList())));
+    }
 }

kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/ListPairTest.java

@@ -0,0 +1,69 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop;
+
+import static org.testng.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.testng.annotations.Test;
+
+public class ListPairTest {
+
+    @Test
+    public void testGetter() {
+        ListPair<Integer> pair = ListPair.of(Collections.emptyMap());
+        assertEquals(pair.getSuccessfulList(), Collections.emptyList());
+        assertEquals(pair.getFailedList(), Collections.emptyList());
+
+        pair = ListPair.of(null);
+        assertEquals(pair.getSuccessfulList(), Collections.emptyList());
+        assertEquals(pair.getFailedList(), Collections.emptyList());
+
+        final Map<Boolean, List<Integer>> data = new HashMap<>();
+        data.put(true, Arrays.asList(1, 2, 3));
+        pair = ListPair.of(data);
+        assertEquals(pair.getSuccessfulList(), Arrays.asList(1, 2, 3));
+        assertEquals(pair.getFailedList(), Collections.emptyList());
+
+        data.remove(true);
+        data.put(false, Arrays.asList(4, 5));
+        pair = ListPair.of(data);
+        assertEquals(pair.getSuccessfulList(), Collections.emptyList());
+        assertEquals(pair.getFailedList(), Arrays.asList(4, 5));
+
+        data.put(true, Arrays.asList(6, 7));
+        assertEquals(pair.getSuccessfulList(), Arrays.asList(6, 7));
+        assertEquals(pair.getFailedList(), Arrays.asList(4, 5));
+    }
+
+    @Test
+    public void testSplit() {
+        final ListPair<Integer> pair = ListPair.split(Stream.of(1, 2, 3, 4, 5), i -> i % 2 == 0);
+        assertEquals(pair.getSuccessfulList(), Arrays.asList(2, 4));
+        assertEquals(pair.getFailedList(), Arrays.asList(1, 3, 5));
+    }
+
+    @Test
+    public void testMap() {
+        final ListPair<String> pair = ListPair.split(Stream.of(1, 2, 3), i -> i % 2 == 0)
+                .map(i -> "msg-" + i);
+        assertEquals(pair.getSuccessfulList(), Collections.singletonList("msg-2"));
+        assertEquals(pair.getFailedList(), Arrays.asList("msg-1", "msg-3"));
+    }
+}

kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop.utils;
+
+import static org.testng.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import org.testng.annotations.Test;
+
+public class CoreUtilsTest {
+
+    @Test
+    public void testListOperations() {
+        final List<Integer> integers = Arrays.asList(1, 2, 3);
+        final List<String> strings = CoreUtils.listToList(integers, i -> "msg-" + i);
+        assertEquals(strings, Arrays.asList("msg-1", "msg-2", "msg-3"));
+
+        final Map<Integer, String> map = CoreUtils.listToMap(integers, i -> "value-" + i);
+        final Map<Integer, String> expectedMap = new HashMap<>();
+        expectedMap.put(1, "value-1");
+        expectedMap.put(2, "value-2");
+        expectedMap.put(3, "value-3");
+        assertEquals(map, expectedMap);
+
+        assertEquals(
+                CoreUtils.mapToList(map, (i, s) -> i + "-" + s),
+                Arrays.asList("1-value-1", "2-value-2", "3-value-3")
+        );
+    }
+
+    @Test
+    public void testWaitForAll() throws ExecutionException, InterruptedException {
+        final List<CompletableFuture<Integer>> futures = Arrays.asList(
+                CompletableFuture.completedFuture(1),
+                CompletableFuture.completedFuture(2),
+                CompletableFuture.completedFuture(3));
+        final List<String> strings = CoreUtils.waitForAll(futures,
+                integers -> CoreUtils.listToList(integers, Object::toString)).get();
+        assertEquals(strings, Arrays.asList("1", "2", "3"));
+    }
+}

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerWithAuthorizationTest.java

@@ -241,7 +241,7 @@ public void testMetadataListTopic() throws Exception {
 
         final RequestHeader header = new RequestHeader(ApiKeys.METADATA, (short) 1, "client", 0);
         final MetadataRequest request =
-                new MetadataRequest(Collections.emptyList(), true, (short) 1);
+                new MetadataRequest(Collections.emptyList(), true, (short) 0);
         final CompletableFuture<AbstractResponse> responseFuture = new CompletableFuture<>();
         spyHandler.handleTopicMetadataRequest(
                 new KafkaCommandDecoder.KafkaHeaderAndRequest(