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..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 @@ -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; @@ -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; @@ -140,7 +141,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 +167,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 +464,45 @@ 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 { + log.warn("[{}] Failed to create partitioned topic {}: {}", + ctx.channel(), topicName, createException.getMessage()); + 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 +535,168 @@ 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); - } - }; - 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; + 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()); }); - return topicMapFuture; - } - - @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()); + if (topicToPartitionIndexes.isEmpty()) { + return Collections.emptyList(); } - // 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<>(); + // 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 { + // 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; + } + + 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()), + topics -> topics.stream().flatMap(List::stream)); + } + + private CompletableFuture> authorizeTopicsAsync(final Collection topics, + final AclOperation aclOperation) { + final Map> futureMap = topics.stream().collect( + Collectors.toMap( + topic -> topic, + topic -> authorize(aclOperation, Resource.of(ResourceType.TOPIC, topic)) + )); + return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> + ListPair.of(futureMap.entrySet() + .stream() + .collect(Collectors.groupingBy(e -> e.getValue().join())) + ).map(Map.Entry::getKey)); + } + + private CompletableFuture> findTopicMetadata(final ListPair listPair, + final boolean allowTopicAutoCreation) { + final Map> futureMap = CoreUtils.listToMap( + listPair.getSuccessfulList(), + topic -> getPartitionedTopicMetadataAsync(topic, allowTopicAutoCreation) + ); + return CoreUtils.waitForAll(futureMap.values()).thenApply(__ -> + CoreUtils.mapToList(futureMap, (key, value) -> new TopicAndMetadata(key, value.join())) + ).thenApply(authorizedTopicAndMetadataList -> + ListUtils.union(authorizedTopicAndMetadataList, + CoreUtils.listToList(listPair.getFailedList(), + topic -> new TopicAndMetadata(topic, TopicAndMetadata.AUTHORIZATION_FAILURE)) + ) + ); + } - 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) -> { - 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); - 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); + 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 + )); + // 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) -> { + if (e != null) { + log.error("[{}] Request {}: Exception fetching metadata", ctx.channel(), metadataHar.getHeader(), e); + resultFuture.completeExceptionally(e); 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 ListPair listPair = + ListPair.split(topicAndMetadataList.stream(), TopicAndMetadata::hasNoError); + CoreUtils.waitForAll(listPair.getSuccessfulList().stream() + .map(topicAndMetadata -> + topicAndMetadata.lookupAsync(this::findBroker, getOriginalTopic, metadataNamespace) + ).collect(Collectors.toList()), successfulTopicMetadataList -> { + final List topicMetadataList = ListUtils.union(successfulTopicMetadataList, + CoreUtils.listToList(listPair.getFailedList(), + metadata -> metadata.toTopicMetadata(getOriginalTopic, metadataNamespace)) + ); + 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 +884,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 +2359,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/ListPair.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ListPair.java new file mode 100644 index 0000000000..37f32b9e31 --- /dev/null +++ b/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>, 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(CoreUtils.mapValue(data, list -> CoreUtils.listToList(list, function))); + } + + 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(); + } +} 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..6b79077d00 --- /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()), partitionMetadataList -> + new TopicMetadata( + error(), + getOriginalTopic.apply(topic), + KopTopic.isInternalTopic(topic, metadataNamespace), + partitionMetadataList + )); + } + + 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..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 @@ -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 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])); + } + + public static CompletableFuture waitForAll(final Collection> futures, + final Function, R> function) { + return waitForAll(futures).thenApply(__ -> + function.apply(futures.stream().map(CompletableFuture::join).collect(Collectors.toList()))); + } } 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..ed7cc3b948 --- /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 testListOperations() { + 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")); + } +} 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(