From 48d8210aeee0974e99abfb96902bfd72ba85d065 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 01:09:50 +0800 Subject: [PATCH 01/11] Refactor the handler for METADATA request --- .../handlers/kop/AuthorizedTopicsPair.java | 33 ++ .../handlers/kop/KafkaRequestHandler.java | 542 +++++++----------- .../pulsar/handlers/kop/TopicAndMetadata.java | 98 ++++ .../pulsar/handlers/kop/utils/CoreUtils.java | 11 + 4 files changed, 349 insertions(+), 335 deletions(-) create mode 100644 kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java create mode 100644 kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java new file mode 100644 index 0000000000..9c0b5a2457 --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java @@ -0,0 +1,33 @@ +/** + * 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 java.util.List; +import java.util.Map; +import lombok.AllArgsConstructor; + +@AllArgsConstructor +public class AuthorizedTopicsPair { + + // Use map as its underlying type because it can be retrieved from Collectors.groupingBy(). + private final Map> data; + + public List getAuthorizedTopics() { + return data.get(true); + } + + public List getUnauthorizedTopics() { + return data.get(false); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index fbcad33bcb..c840d8cb12 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -22,7 +22,6 @@ import static org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails; import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.common.collect.Sets; import io.netty.buffer.ByteBuf; @@ -77,8 +76,9 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; import java.util.function.Consumer; +import java.util.function.Function; import java.util.stream.Collectors; -import java.util.stream.IntStream; +import java.util.stream.Stream; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.util.OrderedScheduler; @@ -140,7 +140,6 @@ import org.apache.kafka.common.requests.ListOffsetRequest; import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.requests.MetadataRequest; -import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata; import org.apache.kafka.common.requests.MetadataResponse.TopicMetadata; import org.apache.kafka.common.requests.OffsetCommitRequest; @@ -167,7 +166,6 @@ import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.Murmur3_32Hash; @@ -465,8 +463,43 @@ protected void handleInactive(KafkaHeaderAndRequest kafkaHeaderAndRequest, } // Leverage pulsar admin to get partitioned topic metadata - private CompletableFuture getPartitionedTopicMetadataAsync(String topicName) { - return admin.topics().getPartitionedTopicMetadataAsync(topicName); + // NOTE: the returned future never completes exceptionally + private CompletableFuture getPartitionedTopicMetadataAsync(String topicName, + boolean allowAutoTopicCreation) { + final CompletableFuture future = new CompletableFuture<>(); + admin.topics().getPartitionedTopicMetadataAsync(topicName).whenComplete((metadata, e) -> { + if (e == null) { + if (metadata.partitions > 0) { + if (log.isDebugEnabled()) { + log.debug("Topic {} has {} partitions", topicName, metadata.partitions); + } + future.complete(metadata.partitions); + } else { + future.complete(TopicAndMetadata.NON_PARTITIONED_NUMBER); + } + } else if (e instanceof PulsarAdminException.NotFoundException) { + if (allowAutoTopicCreation) { + log.info("[{}] Topic {} doesn't exist, auto create it with {} partitions", + ctx.channel(), topicName, defaultNumPartitions); + admin.topics().createPartitionedTopicAsync(topicName, defaultNumPartitions) + .whenComplete((__, createException) -> { + if (createException == null) { + future.complete(defaultNumPartitions); + } else { + future.complete(TopicAndMetadata.INVALID_PARTITIONS); + } + }); + } else { + log.error("[{}] Topic {} doesn't exist and it's not allowed to auto create partitioned topic", + ctx.channel(), topicName, e); + future.complete(TopicAndMetadata.INVALID_PARTITIONS); + } + } else { + log.error("[{}] Failed to get partitioned topic {}", ctx.channel(), topicName, e); + future.complete(TopicAndMetadata.INVALID_PARTITIONS); + } + }); + return future; } private CompletableFuture> expandAllowedNamespaces(Set allowedNamespaces) { @@ -499,348 +532,177 @@ static CompletableFuture> expandAllowedNamespaces(Set allowe .thenApply(f -> result); } - // Get all topics in the configured allowed namespaces. - // key: the full topic name without partition suffix, e.g. persistent://public/default/my-topic - // value: the partitions associated with the key, e.g. for a topic with 3 partitions, - // persistent://public/default/my-topic-partition-0 - // persistent://public/default/my-topic-partition-1 - // persistent://public/default/my-topic-partition-2 - private CompletableFuture>> getAllTopicsAsync() { - CompletableFuture>> topicMapFuture = new CompletableFuture<>(); - final Map> topicMap = new ConcurrentHashMap<>(); - final CompletableFuture> allowedNamespacesFuture = - expandAllowedNamespaces(kafkaConfig.getKopAllowedNamespaces()); - String namespacePrefix = currentNamespacePrefix(); - allowedNamespacesFuture.thenAccept(allowedNamespaces -> { - final AtomicInteger pendingNamespacesCount = new AtomicInteger(allowedNamespaces.size()); - Runnable completeOne = () -> { - if (pendingNamespacesCount.decrementAndGet() == 0) { - topicMapFuture.complete(topicMap); + private List analyzeFullTopicNames(final Stream fullTopicNames) { + // key is the topic name, value is a list of the topic's partition indexes + final Map> topicToPartitionIndexes = new HashMap<>(); + fullTopicNames.forEach(fullTopicName -> { + final TopicName topicName = TopicName.get(fullTopicName); + topicToPartitionIndexes.computeIfAbsent( + topicName.getPartitionedTopicName(), + ignored -> new ArrayList<>() + ).add(topicName.getPartitionIndex()); + }); + if (topicToPartitionIndexes.isEmpty()) { + return Collections.emptyList(); + } + + // Check missed partitions + final List topicAndMetadataList = new ArrayList<>(); + topicToPartitionIndexes.forEach((topic, partitionIndexes) -> { + Collections.sort(partitionIndexes); + final int lastIndex = partitionIndexes.get(partitionIndexes.size() - 1); + if (lastIndex < 0) { + topicAndMetadataList.add( + new TopicAndMetadata(topic, TopicAndMetadata.NON_PARTITIONED_NUMBER)); + } else { + if (lastIndex == partitionIndexes.size() - 1) { + topicAndMetadataList.add(new TopicAndMetadata(topic, partitionIndexes.size())); + } else { + log.warn("The partitions of topic {} is wrong ({}), try to create missed partitions", + topic, partitionIndexes.size()); + admin.topics().createMissedPartitionsAsync(topic); } - }; - for (String namespace : allowedNamespaces) { - authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.NAMESPACE, namespace)) - .whenComplete((isAuthorized, ex) -> { - if (ex != null) { - log.error("Describe namespace authorize failed, namespace - {}. {}", - namespace, ex.getMessage()); - completeOne.run(); - return; - } - if (isAuthorized) { - pulsarService.getNamespaceService() - .getListOfPersistentTopics(NamespaceName.get(namespace)) - .whenComplete((topics, e) -> { - if (e != null) { - log.error("Failed to getListOfPersistentTopic of {}", namespace, e); - topicMapFuture.completeExceptionally(e); - return; - } - if (topicMapFuture.isCompletedExceptionally()) { - return; - } - for (String topic : topics) { - final TopicName topicName = TopicName.get(topic); - final String key = topicName.getPartitionedTopicName(); - topicMap.computeIfAbsent( - KopTopic.removeDefaultNamespacePrefix(key, namespacePrefix), - ignored -> Collections.synchronizedList(new ArrayList<>()) - ).add(topicName); - } - completeOne.run(); - }); - } else { - completeOne.run(); - } - }); } - }).exceptionally(error -> { - topicMapFuture.completeExceptionally(error); - return null; }); - return topicMapFuture; + return topicAndMetadataList; + } + + private CompletableFuture> authorizeNamespacesAsync(final Collection namespaces, + final AclOperation aclOperation) { + final Map> futureMap = namespaces.stream().collect( + Collectors.toMap( + namespace -> namespace, + namespace -> authorize(aclOperation, Resource.of(ResourceType.NAMESPACE, namespace)) + )); + return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> + futureMap.entrySet().stream().filter(e -> { + if (!e.getValue().join()) { + log.warn("Failed to authorize {} for ACL operation {}", e.getKey(), aclOperation); + return false; + } + return true; + }).map(Map.Entry::getKey).collect(Collectors.toList()) + ); + } + + private CompletableFuture> listAllTopicsFromNamespacesAsync(final List namespaces) { + return CoreUtils.waitForAll(namespaces.stream() + .map(namespace -> pulsarService.getNamespaceService() + .getListOfPersistentTopics(NamespaceName.get(namespace)) + ).collect(Collectors.toList()), + topicsStream -> topicsStream.flatMap(List::stream)); + } + + private CompletableFuture authorizeTopicsAsync(final Collection topics, + final AclOperation aclOperation) { + final Map> futureMap = topics.stream().collect( + Collectors.toMap( + namespace -> namespace, + namespace -> authorize(aclOperation, Resource.of(ResourceType.NAMESPACE, namespace)) + )); + return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> + new AuthorizedTopicsPair(futureMap.entrySet().stream() + .collect(Collectors.groupingBy(e -> e.getValue().join())) + .entrySet().stream() + .collect(Collectors.toMap( + Map.Entry::getKey, + e -> e.getValue().stream().map(Map.Entry::getKey).collect(Collectors.toList()) + )) + ) + ); } - @Override - protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, - CompletableFuture resultFuture) { - checkArgument(metadataHar.getRequest() instanceof MetadataRequest); - - MetadataRequest metadataRequest = (MetadataRequest) metadataHar.getRequest(); - if (log.isDebugEnabled()) { - log.debug("[{}] Request {}: for topic {} ", - ctx.channel(), metadataHar.getHeader(), metadataRequest.topics()); - } - - // Command response for all topics - List allTopicMetadata = Collections.synchronizedList(Lists.newArrayList()); - List allNodes = Collections.synchronizedList(Lists.newArrayList()); - // Get all kop brokers in local cache - allNodes.addAll(adminManager.getBrokers(advertisedEndPoint.getListenerName())); - - List topics = metadataRequest.topics(); - // topics in format : persistent://%s/%s/abc-partition-x, will be grouped by as: - // Entry - - // A future for a map from to : - // e.g. - // 1. no topics provided, get all topics from namespace; - // 2. topics provided, get provided topics. - CompletableFuture>> pulsarTopicsFuture; - - // Map for , use for findBroker - // e.g. - final Map nonPartitionedTopicMap = Maps.newConcurrentMap(); - - final String metadataNamespace = kafkaConfig.getKafkaMetadataNamespace(); - pulsarTopicsFuture = new CompletableFuture<>(); + private CompletableFuture> findTopicMetadata(final AuthorizedTopicsPair result, + final boolean allowTopicAutoCreation) { + final Map> futureMap = result.getAuthorizedTopics().stream() + .collect(Collectors.toMap( + topic -> topic, + topic -> getPartitionedTopicMetadataAsync(topic, allowTopicAutoCreation)) + ); + return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> + futureMap.entrySet().stream() + .map(e -> new TopicAndMetadata(e.getKey(), e.getValue().join())).collect(Collectors.toList()) + ).thenApply(topicAndMetadataList -> { + result.getUnauthorizedTopics().forEach(topic -> + topicAndMetadataList.add(new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE))); + return topicAndMetadataList; + }); + } - if (topics == null || topics.isEmpty()) { + private CompletableFuture> getTopicsAsync(MetadataRequest request, + Set fullTopicNames) { + // The implementation of MetadataRequest#isAllTopics() in kafka-clients 2.0 is wrong. + // Because in version 0, an empty topic list indicates "request metadata for all topics." + if ((request.topics() == null) || (request.topics().isEmpty() && request.version() == 0)) { // clean all cache when get all metadata for librdkafka(<1.0.0). KopBrokerLookupManager.clear(); - // get all topics, filter by permissions. - final Map> topicMap = new ConcurrentHashMap<>(); - getAllTopicsAsync().thenAccept((allTopicMap) -> { - allTopicMap.forEach((topic, list) -> - list.forEach((topicName -> - topicMap.computeIfAbsent(topic, ignored -> new ArrayList<>()).add(topicName)) - ) - ); - pulsarTopicsFuture.complete(allTopicMap); - }); + return expandAllowedNamespaces(kafkaConfig.getKopAllowedNamespaces()) + .thenCompose(namespaces -> authorizeNamespacesAsync(namespaces, AclOperation.DESCRIBE)) + .thenCompose(this::listAllTopicsFromNamespacesAsync) + .thenApply(this::analyzeFullTopicNames); } else { - // get only the provided topics - final Map> pulsarTopics = Maps.newConcurrentMap(); - - List requestTopics = metadataRequest.topics(); - final int topicsNumber = requestTopics.size(); - AtomicInteger topicsCompleted = new AtomicInteger(0); - - final Runnable completeOneTopic = () -> { - if (topicsCompleted.incrementAndGet() == topicsNumber) { - if (log.isDebugEnabled()) { - log.debug("[{}] Request {}: Completed get {} topic's partitions", - ctx.channel(), metadataHar.getHeader(), topicsNumber); - } - pulsarTopicsFuture.complete(pulsarTopics); - } - }; - final String namespacePrefix = currentNamespacePrefix(); - final BiConsumer addTopicPartition = (topic, partition) -> { - final KopTopic kopTopic = new KopTopic(topic, namespacePrefix); - pulsarTopics.putIfAbsent(topic, - IntStream.range(0, partition) - .mapToObj(i -> TopicName.get(kopTopic.getPartitionName(i))) - .collect(Collectors.toList())); - completeOneTopic.run(); - }; - - final BiConsumer completeOneAuthFailedTopic = (topic, fullTopicName) -> { - allTopicMetadata.add(new TopicMetadata( - Errors.TOPIC_AUTHORIZATION_FAILED, - topic, - KopTopic.isInternalTopic(fullTopicName, metadataNamespace), - Collections.emptyList())); - completeOneTopic.run(); - }; - - requestTopics.forEach(topic -> { - final String fullTopicName = new KopTopic(topic, namespacePrefix).getFullName(); - - authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.TOPIC, fullTopicName)) - .whenComplete((authorized, ex) -> { - if (ex != null) { - log.error("Describe topic authorize failed, topic - {}. {}", - fullTopicName, ex.getMessage()); - // Authentication failed - completeOneAuthFailedTopic.accept(topic, fullTopicName); - return; - } - if (!authorized) { - // Permission denied - completeOneAuthFailedTopic.accept(topic, fullTopicName); - return; - } - // get partition numbers for each topic. - // If topic doesn't exist and allowAutoTopicCreation is enabled, - // the topic will be created first. - getPartitionedTopicMetadataAsync(fullTopicName) - .whenComplete((partitionedTopicMetadata, throwable) -> { - if (throwable != null) { - if (throwable instanceof PulsarAdminException.NotFoundException) { - if (kafkaConfig.isAllowAutoTopicCreation() - && metadataRequest.allowAutoTopicCreation()) { - log.info("[{}] Request {}: Topic {} doesn't exist, " - + "auto create it with {} partitions", - ctx.channel(), metadataHar.getHeader(), - topic, defaultNumPartitions); - admin.topics().createPartitionedTopicAsync( - fullTopicName, defaultNumPartitions) - .whenComplete((ignored, e) -> { - if (e == null) { - addTopicPartition.accept(topic, defaultNumPartitions); - } else { - log.error("[{}] Failed to create partitioned topic {}", - ctx.channel(), topic, e); - completeOneTopic.run(); - } - }); - } else { - log.error("[{}] Request {}: Topic {} doesn't exist and it's " - + "not allowed to auto create partitioned topic", - ctx.channel(), metadataHar.getHeader(), topic); - // not allow to auto create topic, return unknown topic - allTopicMetadata.add( - new TopicMetadata( - Errors.UNKNOWN_TOPIC_OR_PARTITION, - topic, - KopTopic.isInternalTopic(fullTopicName, - metadataNamespace), - Collections.emptyList())); - completeOneTopic.run(); - } - } else { - // Failed get partitions. - allTopicMetadata.add( - new TopicMetadata( - Errors.UNKNOWN_TOPIC_OR_PARTITION, - topic, - KopTopic.isInternalTopic(fullTopicName, metadataNamespace), - Collections.emptyList())); - log.warn("[{}] Request {}: Failed to get partitioned pulsar topic {} " - + "metadata: {}", - ctx.channel(), metadataHar.getHeader(), - fullTopicName, throwable.getMessage()); - completeOneTopic.run(); - } - } else { // the topic already existed - if (partitionedTopicMetadata.partitions > 0) { - if (log.isDebugEnabled()) { - log.debug("Topic {} has {} partitions", - topic, partitionedTopicMetadata.partitions); - } - addTopicPartition.accept(topic, partitionedTopicMetadata.partitions); - } else { - // In case non-partitioned topic, treat as a one partitioned topic. - nonPartitionedTopicMap.put(TopicName - .get(fullTopicName) - .getPartition(0) - .toString(), - TopicName.get(fullTopicName) - ); - addTopicPartition.accept(topic, 1); - } - } - }); - }); - - }); + final boolean allowTopicCreation = kafkaConfig.isAllowAutoTopicCreation() + && request.allowAutoTopicCreation(); + return authorizeTopicsAsync(fullTopicNames, AclOperation.DESCRIBE) + .thenCompose(authorizedTopicsPair -> findTopicMetadata(authorizedTopicsPair, allowTopicCreation)); } + } + + @Override + protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, + CompletableFuture resultFuture) { + // Get all kop brokers in local cache + List allNodes = Collections.synchronizedList( + new ArrayList<>(adminManager.getBrokers(advertisedEndPoint.getListenerName()))); - // 2. After get all topics, for each topic, get the service Broker for it, and add to response - AtomicInteger topicsCompleted = new AtomicInteger(0); // Each Pulsar broker can manage metadata like controller in Kafka, // Kafka's AdminClient needs to find a controller node for metadata management. // So here we return an random broker as a controller for the given listenerName. final int controllerId = adminManager.getControllerId(advertisedEndPoint.getListenerName()); - pulsarTopicsFuture.whenComplete((pulsarTopics, e) -> { + + final String namespacePrefix = currentNamespacePrefix(); + final MetadataRequest request = (MetadataRequest) metadataHar.getRequest(); + // This map is used to find the original topic name. Both key and value don't have the "-partition-" suffix. + final Map fullTopicNameToOriginal = (request.topics() == null) + ? Collections.emptyMap() + : request.topics().stream().collect( + Collectors.toMap( + topic -> new KopTopic(topic, namespacePrefix).getFullName(), + topic -> topic + )); + final Function getOriginalTopic = fullTopicName -> + fullTopicNameToOriginal.getOrDefault(fullTopicName, fullTopicName); + + final String metadataNamespace = kafkaConfig.getKafkaMetadataNamespace(); + getTopicsAsync(request, fullTopicNameToOriginal.keySet()).whenComplete((topicAndMetadataList, e) -> { if (e != null) { - log.warn("[{}] Request {}: Exception fetching metadata, will return null Response", - ctx.channel(), metadataHar.getHeader(), e); - MetadataResponse finalResponse = - KafkaResponseUtils.newMetadata( - allNodes, - clusterName, - controllerId, - Collections.emptyList()); - resultFuture.complete(finalResponse); + log.error("[{}] Request {}: Exception fetching metadata, will return null Response", + ctx.channel(), metadataHar.getHeader(), e); + resultFuture.complete(KafkaResponseUtils.newMetadata( + allNodes, + clusterName, + controllerId, + Collections.emptyList())); return; } - final int topicsNumber = pulsarTopics.size(); - - if (topicsNumber == 0) { - // no topic partitions added, return now. - MetadataResponse finalResponse = - KafkaResponseUtils.newMetadata( - allNodes, - clusterName, - controllerId, - allTopicMetadata); - resultFuture.complete(finalResponse); - return; - } - final String namespacePrefix = currentNamespacePrefix(); - pulsarTopics.forEach((topic, list) -> { - final int partitionsNumber = list.size(); - AtomicInteger partitionsCompleted = new AtomicInteger(0); - List partitionMetadatas = Collections - .synchronizedList(Lists.newArrayListWithExpectedSize(partitionsNumber)); - - list.forEach(topicName -> { - // For non-partitioned topic. - TopicName realTopicName = nonPartitionedTopicMap.getOrDefault(topicName.toString(), topicName); - findBroker(realTopicName) - .whenComplete(((partitionMetadata, throwable) -> { - if (throwable != null || partitionMetadata == null) { - log.warn("[{}] Request {}: Exception while find Broker metadata", - ctx.channel(), metadataHar.getHeader(), throwable); - partitionMetadatas.add(newFailedPartitionMetadata(realTopicName)); - } else { - Node newNode = partitionMetadata.leader(); - synchronized (allNodes) { - if (!allNodes.stream().anyMatch(node1 -> node1.equals(newNode))) { - allNodes.add(newNode); - } - } - partitionMetadatas.add(partitionMetadata); - } - - // whether completed this topic's partitions list. - int finishedPartitions = partitionsCompleted.incrementAndGet(); - if (log.isDebugEnabled()) { - log.debug("[{}] Request {}: FindBroker for topic {}, partitions found/all: {}/{}.", - ctx.channel(), metadataHar.getHeader(), - topic, finishedPartitions, partitionsNumber); - } - if (finishedPartitions == partitionsNumber) { - // new TopicMetadata for this topic - allTopicMetadata.add( - new TopicMetadata( - Errors.NONE, - // The topic returned to Kafka clients should be - // the same with what it sent - topic, - KopTopic.isInternalTopic( - new KopTopic(topic, namespacePrefix).getFullName(), - metadataNamespace), - partitionMetadatas)); - - // whether completed all the topics requests. - int finishedTopics = topicsCompleted.incrementAndGet(); - if (log.isDebugEnabled()) { - log.debug("[{}] Request {}: Completed findBroker for topic {}, " - + "partitions found/all: {}/{}. \n dump All Metadata:", - ctx.channel(), metadataHar.getHeader(), topic, - finishedTopics, topicsNumber); - - allTopicMetadata.stream() - .forEach(data -> log.debug("TopicMetadata response: {}", - data.toString())); - } - if (finishedTopics == topicsNumber) { - // TODO: confirm right value for controller_id - MetadataResponse finalResponse = - KafkaResponseUtils.newMetadata( - allNodes, - clusterName, - controllerId, - allTopicMetadata); - resultFuture.complete(finalResponse); - } - } - })); - }); + final Map> topicAndMetadataMap = topicAndMetadataList.stream() + .collect(Collectors.groupingBy(TopicAndMetadata::hasNoError)); + CoreUtils.waitForAll(topicAndMetadataMap.get(true).stream() + .map(topicAndMetadata -> + topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace)) + .collect(Collectors.toList()), topicMetadataStream -> { + final List topicMetadataList = topicAndMetadataMap.get(false).stream() + .map(metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) + .collect(Collectors.toList()); + topicMetadataStream.forEach(topicMetadataList::add); + resultFuture.complete( + KafkaResponseUtils.newMetadata(allNodes, clusterName, controllerId, topicMetadataList)); + return null; + }).exceptionally(lookupException -> { + log.error("[{}] Unexpected exception during lookup", ctx.channel(), lookupException); + resultFuture.completeExceptionally(lookupException); + return null; }); }); } @@ -1028,10 +890,10 @@ protected void handleFindCoordinatorRequest(KafkaHeaderAndRequest findCoordinato // Store group name to metadata store for current client. storeGroupId(groupId, groupIdPath) .thenAccept(__ -> findBroker(TopicName.get(pulsarTopicName)) - .whenComplete((node, t) -> { - if (t != null || node == null){ - log.error("[{}] Request {}: Error while find coordinator, .", - ctx.channel(), findCoordinator.getHeader(), t); + .thenAccept(node -> { + if (node.error() != Errors.NONE) { + log.error("[{}] Request {}: Error while find coordinator.", + ctx.channel(), findCoordinator.getHeader()); resultFuture.complete(KafkaResponseUtils .newFindCoordinator(Errors.LEADER_NOT_AVAILABLE)); @@ -2503,15 +2365,25 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { this.close(); } + // The returned future never completes exceptionally public CompletableFuture findBroker(TopicName topic) { if (log.isDebugEnabled()) { log.debug("[{}] Handle Lookup for {}", ctx.channel(), topic); } - return kopBrokerLookupManager.findBroker(topic.toString(), advertisedEndPoint) + final CompletableFuture future = new CompletableFuture<>(); + kopBrokerLookupManager.findBroker(topic.toString(), advertisedEndPoint) .thenApply(listenerInetSocketAddressOpt -> listenerInetSocketAddressOpt .map(inetSocketAddress -> newPartitionMetadata(topic, newNode(inetSocketAddress))) .orElse(null) - ); + ).whenComplete((partitionMetadata, e) -> { + if (e == null || partitionMetadata == null) { + log.warn("[{}] Request {}: Exception while find Broker metadata", ctx.channel(), e); + future.complete(newFailedPartitionMetadata(topic)); + } else { + future.complete(partitionMetadata); + } + }); + return future; } static Node newNode(InetSocketAddress address) { diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java new file mode 100644 index 0000000000..0f02a990e3 --- /dev/null +++ b/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 lookupAsync( + final Function> lookupFunction, + final Function getOriginalTopic, + final String metadataNamespace) { + return CoreUtils.waitForAll(stream() + .map(TopicName::get) + .map(lookupFunction) + .collect(Collectors.toList()), partitionMetadataStream -> + new TopicMetadata( + error(), + getOriginalTopic.apply(topic), + KopTopic.isInternalTopic(topic, metadataNamespace), + partitionMetadataStream.collect(Collectors.toList()) + )); + } + + public Stream stream() { + if (numPartitions > 0) { + return IntStream.range(0, numPartitions) + .mapToObj(i -> topic + "-partition-" + i); + } else { + return Stream.of(topic); + } + } + + public TopicMetadata toTopicMetadata(final Function getOriginalTopic, + final String metadataNamespace) { + return new TopicMetadata( + error(), + getOriginalTopic.apply(topic), + KopTopic.isInternalTopic(topic, metadataNamespace), + Collections.emptyList() + ); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index 255a0a75cc..010b853783 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -13,14 +13,17 @@ */ package io.streamnative.pulsar.handlers.kop.utils; +import java.util.Collection; 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.Function; import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; +import java.util.stream.Stream; import lombok.experimental.UtilityClass; /** @@ -76,4 +79,12 @@ public static Map mapKeyValue(Map map, )); } + public static CompletableFuture waitForAll(final Collection> futures) { + return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); + } + + public static CompletableFuture waitForAll(final Collection> futures, + final Function, R> function) { + return waitForAll(futures).thenApply(__ -> function.apply(futures.stream().map(CompletableFuture::join))); + } } From 3bdced9c3542769fa6a5975e8cdce64ca28707ac Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 14:56:59 +0800 Subject: [PATCH 02/11] Fix bugs in findBroker and NPE --- .../handlers/kop/AuthorizedTopicsPair.java | 33 ---------- .../handlers/kop/KafkaRequestHandler.java | 52 +++++++-------- .../pulsar/handlers/kop/utils/CoreUtils.java | 17 +++++ .../pulsar/handlers/kop/utils/ListPair.java | 64 +++++++++++++++++++ 4 files changed, 105 insertions(+), 61 deletions(-) delete mode 100644 kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java create mode 100644 kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java deleted file mode 100644 index 9c0b5a2457..0000000000 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AuthorizedTopicsPair.java +++ /dev/null @@ -1,33 +0,0 @@ -/** - * 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 java.util.List; -import java.util.Map; -import lombok.AllArgsConstructor; - -@AllArgsConstructor -public class AuthorizedTopicsPair { - - // Use map as its underlying type because it can be retrieved from Collectors.groupingBy(). - private final Map> data; - - public List getAuthorizedTopics() { - return data.get(true); - } - - public List getUnauthorizedTopics() { - return data.get(false); - } -} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index c840d8cb12..896479cfc8 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -48,6 +48,7 @@ import io.streamnative.pulsar.handlers.kop.utils.KafkaRequestUtils; import io.streamnative.pulsar.handlers.kop.utils.KafkaResponseUtils; import io.streamnative.pulsar.handlers.kop.utils.KopTopic; +import io.streamnative.pulsar.handlers.kop.utils.ListPair; import io.streamnative.pulsar.handlers.kop.utils.MessageMetadataUtils; import io.streamnative.pulsar.handlers.kop.utils.MetadataUtils; import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder; @@ -593,7 +594,7 @@ private CompletableFuture> listAllTopicsFromNamespacesAsync(final topicsStream -> topicsStream.flatMap(List::stream)); } - private CompletableFuture authorizeTopicsAsync(final Collection topics, + private CompletableFuture> authorizeTopicsAsync(final Collection topics, final AclOperation aclOperation) { final Map> futureMap = topics.stream().collect( Collectors.toMap( @@ -601,31 +602,26 @@ private CompletableFuture authorizeTopicsAsync(final Colle namespace -> authorize(aclOperation, Resource.of(ResourceType.NAMESPACE, namespace)) )); return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> - new AuthorizedTopicsPair(futureMap.entrySet().stream() - .collect(Collectors.groupingBy(e -> e.getValue().join())) - .entrySet().stream() - .collect(Collectors.toMap( - Map.Entry::getKey, - e -> e.getValue().stream().map(Map.Entry::getKey).collect(Collectors.toList()) - )) - ) - ); + ListPair.of(futureMap.entrySet() + .stream() + .collect(Collectors.groupingBy(e -> e.getValue().join())) + ).map(Map.Entry::getKey)); } - private CompletableFuture> findTopicMetadata(final AuthorizedTopicsPair result, + private CompletableFuture> findTopicMetadata(final ListPair listPair, final boolean allowTopicAutoCreation) { - final Map> futureMap = result.getAuthorizedTopics().stream() - .collect(Collectors.toMap( - topic -> topic, - topic -> getPartitionedTopicMetadataAsync(topic, allowTopicAutoCreation)) - ); + final Map> futureMap = CoreUtils.listToMap( + listPair.getSuccessfulList(), + topic -> getPartitionedTopicMetadataAsync(topic, allowTopicAutoCreation) + ); return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> - futureMap.entrySet().stream() - .map(e -> new TopicAndMetadata(e.getKey(), e.getValue().join())).collect(Collectors.toList()) + CoreUtils.mapToList(futureMap, (key, value) -> new TopicAndMetadata(key, value.join())) ).thenApply(topicAndMetadataList -> { - result.getUnauthorizedTopics().forEach(topic -> - topicAndMetadataList.add(new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE))); - return topicAndMetadataList; + // Create a new list in case `topicAndMetadataList` is not modifiable. + final List list = CoreUtils.listToList(listPair.getFailedList(), + topic -> new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE)); + list.addAll(topicAndMetadataList); + return list; }); } @@ -686,13 +682,13 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, return; } - final Map> topicAndMetadataMap = topicAndMetadataList.stream() - .collect(Collectors.groupingBy(TopicAndMetadata::hasNoError)); - CoreUtils.waitForAll(topicAndMetadataMap.get(true).stream() + final ListPair listPair = + ListPair.split(topicAndMetadataList.stream(), TopicAndMetadata::hasNoError); + CoreUtils.waitForAll(listPair.getSuccessfulList().stream() .map(topicAndMetadata -> - topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace)) - .collect(Collectors.toList()), topicMetadataStream -> { - final List topicMetadataList = topicAndMetadataMap.get(false).stream() + topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace) + ).collect(Collectors.toList()), topicMetadataStream -> { + final List topicMetadataList = listPair.getFailedList().stream() .map(metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) .collect(Collectors.toList()); topicMetadataStream.forEach(topicMetadataList::add); @@ -2376,7 +2372,7 @@ public CompletableFuture findBroker(TopicName topic) { .map(inetSocketAddress -> newPartitionMetadata(topic, newNode(inetSocketAddress))) .orElse(null) ).whenComplete((partitionMetadata, e) -> { - if (e == null || partitionMetadata == null) { + if (e != null || partitionMetadata == null) { log.warn("[{}] Request {}: Exception while find Broker metadata", ctx.channel(), e); future.complete(newFailedPartitionMetadata(topic)); } else { diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index 010b853783..331aee2e28 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -14,11 +14,13 @@ 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; @@ -79,6 +81,21 @@ public static Map mapKeyValue(Map map, )); } + public static List listToList(final List list, + final Function function) { + return list.stream().map(function).collect(Collectors.toList()); + } + + public static Map listToMap(final List list, + final Function function) { + return list.stream().collect(Collectors.toMap(key -> key, function)); + } + + public static List mapToList(final Map map, + final BiFunction function) { + return map.entrySet().stream().map(e -> function.apply(e.getKey(), e.getValue())).collect(Collectors.toList()); + } + public static CompletableFuture waitForAll(final Collection> futures) { return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java new file mode 100644 index 0000000000..3a90ebc97d --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java @@ -0,0 +1,64 @@ +/** + * 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 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>, 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 + */ +public class ListPair { + + private final Map> data; + + public ListPair map(final Function function) { + return of(data.entrySet().stream().collect(Collectors.toMap( + Map.Entry::getKey, + e -> e.getValue().stream().map(function).collect(Collectors.toList()) + ))); + } + + public List getSuccessfulList() { + return Optional.ofNullable(data.get(true)).orElse(Collections.emptyList()); + } + + public List getFailedList() { + return Optional.ofNullable(data.get(false)).orElse(Collections.emptyList()); + } + + public static ListPair split(final Stream stream, + final Function function) { + return ListPair.of(stream.collect(Collectors.groupingBy(function))); + } + + public static ListPair of(final Map> data) { + return new ListPair(data); + } + + private ListPair(final Map> data) { + this.data = (data != null) ? data : Collections.emptyMap(); + } +} From e2b42bffebb7e74114d7ba07e7d814f48b129c46 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 16:24:48 +0800 Subject: [PATCH 03/11] Fix authorization failure --- .../pulsar/handlers/kop/KafkaRequestHandler.java | 6 +++--- .../io/streamnative/pulsar/handlers/kop/utils/ListPair.java | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index 896479cfc8..e58e15c7e1 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -595,11 +595,11 @@ private CompletableFuture> listAllTopicsFromNamespacesAsync(final } private CompletableFuture> authorizeTopicsAsync(final Collection topics, - final AclOperation aclOperation) { + final AclOperation aclOperation) { final Map> futureMap = topics.stream().collect( Collectors.toMap( - namespace -> namespace, - namespace -> authorize(aclOperation, Resource.of(ResourceType.NAMESPACE, namespace)) + topic -> topic, + topic -> authorize(aclOperation, Resource.of(ResourceType.TOPIC, topic)) )); return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> ListPair.of(futureMap.entrySet() diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java index 3a90ebc97d..cda20bf328 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java @@ -55,7 +55,7 @@ public static ListPair split(final Stream stream, } public static ListPair of(final Map> data) { - return new ListPair(data); + return new ListPair<>(data); } private ListPair(final Map> data) { From 034f03555f2d22d8cbf0ac7040761e2954bde21c Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 17:43:22 +0800 Subject: [PATCH 04/11] Fix the list topics failure --- .../pulsar/handlers/kop/KafkaRequestHandler.java | 12 +++++++----- .../pulsar/handlers/kop/TopicAndMetadata.java | 4 ++-- .../pulsar/handlers/kop/utils/CoreUtils.java | 6 +++--- 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index e58e15c7e1..bc2b273769 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -591,7 +591,7 @@ private CompletableFuture> listAllTopicsFromNamespacesAsync(final .map(namespace -> pulsarService.getNamespaceService() .getListOfPersistentTopics(NamespaceName.get(namespace)) ).collect(Collectors.toList()), - topicsStream -> topicsStream.flatMap(List::stream)); + topics -> topics.stream().flatMap(List::stream)); } private CompletableFuture> authorizeTopicsAsync(final Collection topics, @@ -666,8 +666,10 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, topic -> new KopTopic(topic, namespacePrefix).getFullName(), topic -> topic )); - final Function getOriginalTopic = fullTopicName -> - fullTopicNameToOriginal.getOrDefault(fullTopicName, fullTopicName); + // NOTE: for all topics METADATA request, remove the default namespace prefix just for backward compatibility. + final Function getOriginalTopic = fullTopicName -> fullTopicNameToOriginal.isEmpty() + ? KopTopic.removeDefaultNamespacePrefix(fullTopicName, namespacePrefix) + : fullTopicNameToOriginal.getOrDefault(fullTopicName, fullTopicName); final String metadataNamespace = kafkaConfig.getKafkaMetadataNamespace(); getTopicsAsync(request, fullTopicNameToOriginal.keySet()).whenComplete((topicAndMetadataList, e) -> { @@ -687,11 +689,11 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, CoreUtils.waitForAll(listPair.getSuccessfulList().stream() .map(topicAndMetadata -> topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace) - ).collect(Collectors.toList()), topicMetadataStream -> { + ).collect(Collectors.toList()), successfulTopicMetadataList -> { final List topicMetadataList = listPair.getFailedList().stream() .map(metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) .collect(Collectors.toList()); - topicMetadataStream.forEach(topicMetadataList::add); + topicMetadataList.addAll(successfulTopicMetadataList); resultFuture.complete( KafkaResponseUtils.newMetadata(allNodes, clusterName, controllerId, topicMetadataList)); return null; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java index 0f02a990e3..0857759282 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java @@ -68,12 +68,12 @@ public CompletableFuture lookupAsync( return CoreUtils.waitForAll(stream() .map(TopicName::get) .map(lookupFunction) - .collect(Collectors.toList()), partitionMetadataStream -> + .collect(Collectors.toList()), partitionMetadataList -> new TopicMetadata( error(), getOriginalTopic.apply(topic), KopTopic.isInternalTopic(topic, metadataNamespace), - partitionMetadataStream.collect(Collectors.toList()) + partitionMetadataList )); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index 331aee2e28..f01324ae47 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -25,7 +25,6 @@ import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; -import java.util.stream.Stream; import lombok.experimental.UtilityClass; /** @@ -101,7 +100,8 @@ public static CompletableFuture waitForAll(final Collection CompletableFuture waitForAll(final Collection> futures, - final Function, R> function) { - return waitForAll(futures).thenApply(__ -> function.apply(futures.stream().map(CompletableFuture::join))); + final Function, R> function) { + return waitForAll(futures).thenApply(__ -> + function.apply(futures.stream().map(CompletableFuture::join).collect(Collectors.toList()))); } } From d006f17bb682f0de7c93854869298ba5b146c127 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 18:19:42 +0800 Subject: [PATCH 05/11] Add tests for ListPair and CoreUtils --- .../handlers/kop/KafkaRequestHandler.java | 1 - .../handlers/kop/{utils => }/ListPair.java | 8 +-- .../pulsar/handlers/kop/TopicAndMetadata.java | 2 +- .../pulsar/handlers/kop/ListPairTest.java | 69 +++++++++++++++++++ .../handlers/kop/utils/CoreUtilsTest.java | 57 +++++++++++++++ 5 files changed, 130 insertions(+), 7 deletions(-) rename kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/{utils => }/ListPair.java (89%) create mode 100644 kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/ListPairTest.java create mode 100644 kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index bc2b273769..d1bafb823d 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -48,7 +48,6 @@ import io.streamnative.pulsar.handlers.kop.utils.KafkaRequestUtils; import io.streamnative.pulsar.handlers.kop.utils.KafkaResponseUtils; import io.streamnative.pulsar.handlers.kop.utils.KopTopic; -import io.streamnative.pulsar.handlers.kop.utils.ListPair; import io.streamnative.pulsar.handlers.kop.utils.MessageMetadataUtils; import io.streamnative.pulsar.handlers.kop.utils.MetadataUtils; import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ListPair.java similarity index 89% rename from kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java rename to kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ListPair.java index cda20bf328..37f32b9e31 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ListPair.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ListPair.java @@ -11,8 +11,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.streamnative.pulsar.handlers.kop.utils; +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; @@ -35,10 +36,7 @@ public class ListPair { private final Map> data; public ListPair map(final Function function) { - return of(data.entrySet().stream().collect(Collectors.toMap( - Map.Entry::getKey, - e -> e.getValue().stream().map(function).collect(Collectors.toList()) - ))); + return of(CoreUtils.mapValue(data, list -> CoreUtils.listToList(list, function))); } public List getSuccessfulList() { diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java index 0857759282..6b79077d00 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TopicAndMetadata.java @@ -87,7 +87,7 @@ public Stream stream() { } public TopicMetadata toTopicMetadata(final Function getOriginalTopic, - final String metadataNamespace) { + final String metadataNamespace) { return new TopicMetadata( error(), getOriginalTopic.apply(topic), diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/ListPairTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/ListPairTest.java new file mode 100644 index 0000000000..4d50038bdf --- /dev/null +++ b/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 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> 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 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 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")); + } +} diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java new file mode 100644 index 0000000000..5d8d529710 --- /dev/null +++ b/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 testListConversion() { + final List integers = Arrays.asList(1, 2, 3); + final List strings = CoreUtils.listToList(integers, i -> "msg-" + i); + assertEquals(strings, Arrays.asList("msg-1", "msg-2", "msg-3")); + + final Map map = CoreUtils.listToMap(integers, i -> "value-" + i); + final Map 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> futures = Arrays.asList( + CompletableFuture.completedFuture(1), + CompletableFuture.completedFuture(2), + CompletableFuture.completedFuture(3)); + final List strings = CoreUtils.waitForAll(futures, + integers -> CoreUtils.listToList(integers, Object::toString)).get(); + assertEquals(strings, Arrays.asList("1", "2", "3")); + } +} From 3d5d22a0289b5c58912176222a1ce43a867159a4 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 24 Feb 2022 18:28:49 +0800 Subject: [PATCH 06/11] Fix the incorrect METADATA version --- .../handlers/kop/KafkaRequestHandlerWithAuthorizationTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerWithAuthorizationTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerWithAuthorizationTest.java index 1c1e66d17d..8598082b50 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerWithAuthorizationTest.java +++ b/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 responseFuture = new CompletableFuture<>(); spyHandler.handleTopicMetadataRequest( new KafkaCommandDecoder.KafkaHeaderAndRequest( From 9424ea9a4b7016629e27d5f695a997df2d951364 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 25 Feb 2022 11:44:53 +0800 Subject: [PATCH 07/11] Add concatList method to replace addAll() --- .../handlers/kop/KafkaRequestHandler.java | 23 ++++++++++--------- .../pulsar/handlers/kop/utils/CoreUtils.java | 11 +++++++++ .../handlers/kop/utils/CoreUtilsTest.java | 15 ++++++++++++ 3 files changed, 38 insertions(+), 11 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index d1bafb823d..13d8b1ac28 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -486,6 +486,8 @@ private CompletableFuture getPartitionedTopicMetadataAsync(String topic if (createException == null) { future.complete(defaultNumPartitions); } else { + log.warn("[{}] Failed to create partitioned topic {}: {}", + ctx.channel(), topicName, createException.getMessage()); future.complete(TopicAndMetadata.INVALID_PARTITIONS); } }); @@ -615,13 +617,12 @@ private CompletableFuture> findTopicMetadata(final ListPa ); return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> CoreUtils.mapToList(futureMap, (key, value) -> new TopicAndMetadata(key, value.join())) - ).thenApply(topicAndMetadataList -> { - // Create a new list in case `topicAndMetadataList` is not modifiable. - final List list = CoreUtils.listToList(listPair.getFailedList(), - topic -> new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE)); - list.addAll(topicAndMetadataList); - return list; - }); + ).thenApply(authorizedTopicAndMetadataList -> + CoreUtils.concatList(authorizedTopicAndMetadataList, + CoreUtils.listToList(listPair.getFailedList(), + topic -> new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE)) + ) + ); } private CompletableFuture> getTopicsAsync(MetadataRequest request, @@ -689,10 +690,10 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, .map(topicAndMetadata -> topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace) ).collect(Collectors.toList()), successfulTopicMetadataList -> { - final List topicMetadataList = listPair.getFailedList().stream() - .map(metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) - .collect(Collectors.toList()); - topicMetadataList.addAll(successfulTopicMetadataList); + final List topicMetadataList = CoreUtils.concatList(successfulTopicMetadataList, + CoreUtils.listToList(listPair.getFailedList(), + metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) + ); resultFuture.complete( KafkaResponseUtils.newMetadata(allNodes, clusterName, controllerId, topicMetadataList)); return null; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index f01324ae47..d46b2ab61c 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -25,6 +25,7 @@ import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; +import java.util.stream.Stream; import lombok.experimental.UtilityClass; /** @@ -80,6 +81,16 @@ public static Map mapKeyValue(Map map, )); } + public static List concatList(final List list1, final List list2) { + try { + // if `list` can be modifiable, modify the `list1` in place. + list1.addAll(list2); + return list1; + } catch (UnsupportedOperationException ignored) { + return Stream.concat(list1.stream(), list2.stream()).collect(Collectors.toList()); + } + } + public static List listToList(final List list, final Function function) { return list.stream().map(function).collect(Collectors.toList()); diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java index 5d8d529710..135ce54271 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java +++ b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java @@ -14,7 +14,9 @@ package io.streamnative.pulsar.handlers.kop.utils; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -25,6 +27,19 @@ public class CoreUtilsTest { + @Test + public void testListConcat() { + final List list1 = Arrays.asList(1, 2, 3); + final List list2 = Arrays.asList(4, 5); + final List list3 = CoreUtils.concatList(list1, list2); + assertEquals(list3, Arrays.asList(1, 2, 3, 4, 5)); + + final List list4 = new ArrayList<>(list1); + final List list5 = CoreUtils.concatList(list4, list2); + assertEquals(list5, Arrays.asList(1, 2, 3, 4, 5)); + assertSame(list5, list4); // `list4` is modifiable, so the `concatList` happens in place + } + @Test public void testListConversion() { final List integers = Arrays.asList(1, 2, 3); From bdc06ef6075b2a66fb834372a72424a30aca7cf5 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 25 Feb 2022 11:55:43 +0800 Subject: [PATCH 08/11] Change nested if-else to else if --- .../pulsar/handlers/kop/KafkaRequestHandler.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index 13d8b1ac28..bc5a24a21e 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -556,14 +556,14 @@ private List analyzeFullTopicNames(final Stream fullTo if (lastIndex < 0) { topicAndMetadataList.add( new TopicAndMetadata(topic, TopicAndMetadata.NON_PARTITIONED_NUMBER)); + } else if (lastIndex == partitionIndexes.size() - 1) { + topicAndMetadataList.add(new TopicAndMetadata(topic, partitionIndexes.size())); } else { - if (lastIndex == partitionIndexes.size() - 1) { - topicAndMetadataList.add(new TopicAndMetadata(topic, partitionIndexes.size())); - } else { - log.warn("The partitions of topic {} is wrong ({}), try to create missed partitions", - topic, partitionIndexes.size()); - admin.topics().createMissedPartitionsAsync(topic); - } + // The partitions should be [0, 1, ..., n-1], `n` is the number of partitions. If the last index is not + // `n-1`, there must be some missed partitions. + log.warn("The partitions of topic {} is wrong ({}), try to create missed partitions", + topic, partitionIndexes.size()); + admin.topics().createMissedPartitionsAsync(topic); } }); return topicAndMetadataList; From 364f5a89d88ee436a2b5355aa1f865b7e32df5a6 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 25 Feb 2022 13:49:53 +0800 Subject: [PATCH 09/11] Handle the exceptional case well --- .../pulsar/handlers/kop/KafkaRequestHandler.java | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index bc5a24a21e..7df5d3b257 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -674,13 +674,8 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, final String metadataNamespace = kafkaConfig.getKafkaMetadataNamespace(); getTopicsAsync(request, fullTopicNameToOriginal.keySet()).whenComplete((topicAndMetadataList, e) -> { if (e != null) { - log.error("[{}] Request {}: Exception fetching metadata, will return null Response", - ctx.channel(), metadataHar.getHeader(), e); - resultFuture.complete(KafkaResponseUtils.newMetadata( - allNodes, - clusterName, - controllerId, - Collections.emptyList())); + log.error("[{}] Request {}: Exception fetching metadata", ctx.channel(), metadataHar.getHeader(), e); + resultFuture.completeExceptionally(e); return; } From b79b36a560322711754b008617256a1fec79b645 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 25 Feb 2022 17:31:16 +0800 Subject: [PATCH 10/11] Simplify concatList --- .../pulsar/handlers/kop/utils/CoreUtils.java | 8 +------- .../handlers/kop/utils/CoreUtilsTest.java | 18 +++--------------- 2 files changed, 4 insertions(+), 22 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index d46b2ab61c..9fb59041df 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -82,13 +82,7 @@ public static Map mapKeyValue(Map map, } public static List concatList(final List list1, final List list2) { - try { - // if `list` can be modifiable, modify the `list1` in place. - list1.addAll(list2); - return list1; - } catch (UnsupportedOperationException ignored) { - return Stream.concat(list1.stream(), list2.stream()).collect(Collectors.toList()); - } + return Stream.concat(list1.stream(), list2.stream()).collect(Collectors.toList()); } public static List listToList(final List list, diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java index 135ce54271..6f1b612cae 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java +++ b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java @@ -14,9 +14,7 @@ package io.streamnative.pulsar.handlers.kop.utils; import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertSame; -import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -28,20 +26,10 @@ public class CoreUtilsTest { @Test - public void testListConcat() { - final List list1 = Arrays.asList(1, 2, 3); - final List list2 = Arrays.asList(4, 5); - final List list3 = CoreUtils.concatList(list1, list2); - assertEquals(list3, Arrays.asList(1, 2, 3, 4, 5)); + public void testListOperations() { + assertEquals(CoreUtils.concatList(Arrays.asList(1, 2, 3), Arrays.asList(4, 5)), + Arrays.asList(1, 2, 3, 4, 5)); - final List list4 = new ArrayList<>(list1); - final List list5 = CoreUtils.concatList(list4, list2); - assertEquals(list5, Arrays.asList(1, 2, 3, 4, 5)); - assertSame(list5, list4); // `list4` is modifiable, so the `concatList` happens in place - } - - @Test - public void testListConversion() { final List integers = Arrays.asList(1, 2, 3); final List strings = CoreUtils.listToList(integers, i -> "msg-" + i); assertEquals(strings, Arrays.asList("msg-1", "msg-2", "msg-3")); From 8f5bb0fbb715b5b082cefdc84814c82cabd33eb5 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 25 Feb 2022 21:31:01 +0800 Subject: [PATCH 11/11] Use Lists#union instead of implementing our own list concatenation --- .../pulsar/handlers/kop/KafkaRequestHandler.java | 5 +++-- .../io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java | 5 ----- .../pulsar/handlers/kop/utils/CoreUtilsTest.java | 3 --- 3 files changed, 3 insertions(+), 10 deletions(-) diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index 7df5d3b257..5753e77fdf 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -87,6 +87,7 @@ import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.commons.collections4.ListUtils; import org.apache.commons.lang3.NotImplementedException; import org.apache.commons.lang3.tuple.Pair; import org.apache.kafka.clients.admin.NewPartitions; @@ -618,7 +619,7 @@ private CompletableFuture> findTopicMetadata(final ListPa return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> CoreUtils.mapToList(futureMap, (key, value) -> new TopicAndMetadata(key, value.join())) ).thenApply(authorizedTopicAndMetadataList -> - CoreUtils.concatList(authorizedTopicAndMetadataList, + ListUtils.union(authorizedTopicAndMetadataList, CoreUtils.listToList(listPair.getFailedList(), topic -> new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE)) ) @@ -685,7 +686,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, .map(topicAndMetadata -> topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace) ).collect(Collectors.toList()), successfulTopicMetadataList -> { - final List topicMetadataList = CoreUtils.concatList(successfulTopicMetadataList, + final List topicMetadataList = ListUtils.union(successfulTopicMetadataList, CoreUtils.listToList(listPair.getFailedList(), metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) ); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java index 9fb59041df..f01324ae47 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java @@ -25,7 +25,6 @@ import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; -import java.util.stream.Stream; import lombok.experimental.UtilityClass; /** @@ -81,10 +80,6 @@ public static Map mapKeyValue(Map map, )); } - public static List concatList(final List list1, final List list2) { - return Stream.concat(list1.stream(), list2.stream()).collect(Collectors.toList()); - } - public static List listToList(final List list, final Function function) { return list.stream().map(function).collect(Collectors.toList()); diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java index 6f1b612cae..ed7cc3b948 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java +++ b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtilsTest.java @@ -27,9 +27,6 @@ public class CoreUtilsTest { @Test public void testListOperations() { - assertEquals(CoreUtils.concatList(Arrays.asList(1, 2, 3), Arrays.asList(4, 5)), - Arrays.asList(1, 2, 3, 4, 5)); - final List integers = Arrays.asList(1, 2, 3); final List strings = CoreUtils.listToList(integers, i -> "msg-" + i); assertEquals(strings, Arrays.asList("msg-1", "msg-2", "msg-3"));