diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java index cb0b585901..2fe8eeffc3 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java @@ -143,8 +143,16 @@ public void onLoad(NamespaceBundle bundle) { } groupCoordinator.handleGroupImmigration(name.getPartitionIndex()); } - KafkaTopicManager.removeTopicManagerCache(name.toString()); - KopBrokerLookupManager.removeTopicManagerCache(name.toString()); + // deReference topic when unload + KopBrokerLookupManager.removeTopicManagerCache(topic); + KafkaTopicManager.deReference(topic); + + // For non-partitioned topic. + if (!name.isPartitioned()) { + String partitionedZeroTopicName = name.getPartition(0).toString(); + KafkaTopicManager.deReference(partitionedZeroTopicName); + KopBrokerLookupManager.removeTopicManagerCache(partitionedZeroTopicName); + } } } else { log.error("Failed to get owned topic list for " @@ -180,8 +188,16 @@ public void unLoad(NamespaceBundle bundle) { groupCoordinator.handleGroupEmigration(name.getPartitionIndex()); } // deReference topic when unload - KopBrokerLookupManager.removeTopicManagerCache(name.toString()); - KafkaTopicManager.deReference(name.toString()); + KopBrokerLookupManager.removeTopicManagerCache(topic); + KafkaTopicManager.deReference(topic); + + // For non-partitioned topic. + if (!name.isPartitioned()) { + String partitionedZeroTopicName = name.getPartition(0).toString(); + KafkaTopicManager.deReference(partitionedZeroTopicName); + KopBrokerLookupManager.removeTopicManagerCache(partitionedZeroTopicName); + } + } } else { log.error("Failed to get owned topic list for " 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 7628402425..59c5bbed9f 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 @@ -475,6 +475,10 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, // 2. topics provided, get provided topics. CompletableFuture>> pulsarTopicsFuture; + // Map for , use for findBroker + // e.g. + final Map nonPartitionedTopicMap = Maps.newConcurrentMap(); + if (topics == null || topics.isEmpty()) { // clean all cache when get all metadata for librdkafka(<1.0.0). KafkaTopicManager.clearTopicManagerCache(); @@ -505,7 +509,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, } else { pulsarTopicsFuture = new CompletableFuture<>(); // get only the provided topics - final Map> pulsarTopics = new ConcurrentHashMap<>(); + final Map> pulsarTopics = Maps.newConcurrentMap(); List requestTopics = metadataRequest.topics(); final int topicsNumber = requestTopics.size(); @@ -615,14 +619,14 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, } addTopicPartition.accept(topic, partitionedTopicMetadata.partitions); } else { - log.error("Topic {} is a non-partitioned topic", topic); - allTopicMetadata.add( - new TopicMetadata( - Errors.INVALID_TOPIC_EXCEPTION, - topic, - isInternalTopic(fullTopicName), - Collections.emptyList())); - completeOneTopic.run(); + // 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); } } }); @@ -672,63 +676,68 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, List partitionMetadatas = Collections .synchronizedList(Lists.newArrayListWithExpectedSize(partitionsNumber)); - list.forEach(topicName -> - findBroker(topicName) - .whenComplete(((partitionMetadata, throwable) -> { - if (throwable != null || partitionMetadata == null) { - log.warn("[{}] Request {}: Exception while find Broker metadata", - ctx.channel(), metadataHar.getHeader(), throwable); - partitionMetadatas.add(newFailedPartitionMetadata(topicName)); - } else { - Node newNode = partitionMetadata.leader(); - synchronized (allNodes) { - if (!allNodes.stream().anyMatch(node1 -> node1.equals(newNode))) { - allNodes.add(newNode); + 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); } - 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, - isInternalTopic(new KopTopic(topic).getFullName()), - partitionMetadatas)); - - // whether completed all the topics requests. - int finishedTopics = topicsCompleted.incrementAndGet(); + // whether completed this topic's partitions list. + int finishedPartitions = partitionsCompleted.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())); + log.debug("[{}] Request {}: FindBroker for topic {}, partitions found/all: {}/{}.", + ctx.channel(), metadataHar.getHeader(), + topic, finishedPartitions, partitionsNumber); } - if (finishedTopics == topicsNumber) { - // TODO: confirm right value for controller_id - MetadataResponse finalResponse = - new MetadataResponse( - allNodes, - clusterName, - controllerId, - allTopicMetadata); - resultFuture.complete(finalResponse); + 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, + isInternalTopic(new KopTopic(topic).getFullName()), + 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 = + new MetadataResponse( + allNodes, + clusterName, + controllerId, + allTopicMetadata); + resultFuture.complete(finalResponse); + } } - } - }))); + })); + }); }); }); } @@ -2084,7 +2093,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { private CompletableFuture> getProtocolDataToAdvertise(InetSocketAddress pulsarAddress, TopicName topic) { - CompletableFuture> returnFuture = new CompletableFuture<>(); if (pulsarAddress == null) { @@ -2178,7 +2186,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { } if (lookupDataContainsAddress(data, hostAndPort)) { - topicManager.KOP_ADDRESS_CACHE.put(topic.toString(), returnFuture); + KafkaTopicManager.KOP_ADDRESS_CACHE.put(topic.toString(), returnFuture); returnFuture.complete(data.getProtocol(KafkaProtocolHandler.PROTOCOL_NAME)); return; } @@ -2225,8 +2233,8 @@ public CompletableFuture findBroker(TopicName topic) { .thenCompose(address -> getProtocolDataToAdvertise(address, topic)) .whenComplete((stringOptional, throwable) -> { if (!stringOptional.isPresent() || throwable != null) { - log.error("Not get advertise data for Kafka topic:{}. throwable", - topic, throwable); + log.error("Not get advertise data for Kafka topic:{}. throwable: [{}]", + topic, throwable.getMessage()); KafkaTopicManager.removeTopicManagerCache(topic.toString()); returnFuture.complete(null); return; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java index 14ada50cce..8f88556b19 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import lombok.Getter; +import lombok.NonNull; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.BrokerService; @@ -174,8 +175,9 @@ private Producer registerInPersistentTopic(PersistentTopic persistentTopic) { return producer; } - // call pulsarclient.lookup.getbroker to get and own a topic. - // when error happens, the returned future will complete with null. + // call pulsarclient.lookup.getbroker to get and + // own a topic. + // // when error happens, the returned future will complete with null. public CompletableFuture getTopicBroker(String topicName) { if (closed.get()) { if (log.isDebugEnabled()) { @@ -215,35 +217,73 @@ public CompletableFuture> getTopic(String topicName) { } CompletableFuture> topicCompletableFuture = new CompletableFuture<>(); brokerService.getTopicIfExists(topicName).whenComplete((t2, throwable) -> { + TopicName topicNameObject = TopicName.get(topicName); if (throwable != null) { - // The ServiceUnitNotReadyException is retriable so we should print a warning log instead of error log - if (throwable instanceof BrokerServiceException.ServiceUnitNotReadyException) { - log.warn("[{}] Failed to getTopic {}: {}", - requestHandler.ctx.channel(), topicName, throwable.getMessage()); - topicCompletableFuture.complete(Optional.empty()); + // Failed to getTopic from current broker, remove cache, which added in getTopicBroker. + removeTopicManagerCache(topicName); + if (topicNameObject.getPartitionIndex() == 0) { + log.warn("Get partition-0 error [{}].", throwable.getMessage()); } else { - log.error("[{}] Failed to getTopic {}. exception:", - requestHandler.ctx.channel(), topicName, throwable); - topicCompletableFuture.completeExceptionally(throwable); + handleGetTopicException(topicName, topicCompletableFuture, throwable); + return; } - // failed to getTopic from current broker, remove cache, which added in getTopicBroker. - removeTopicManagerCache(topicName); + } + if (t2 != null && t2.isPresent()) { + topicCompletableFuture.complete(Optional.of((PersistentTopic) t2.get())); return; } - if (t2.isPresent()) { - PersistentTopic persistentTopic = (PersistentTopic) t2.get(); - topicCompletableFuture.complete(Optional.of(persistentTopic)); - } else { - log.error("[{}]Get empty topic for name {}", requestHandler.ctx.channel(), topicName); - removeTopicManagerCache(topicName); - topicCompletableFuture.complete(Optional.empty()); + // Fallback try use non-partitioned topic + if (topicNameObject.getPartitionIndex() == 0) { + String nonPartitionedTopicName = topicNameObject.getPartitionedTopicName(); + if (log.isDebugEnabled()) { + log.debug("[{}]Try to get non-partitioned topic for name {}", + requestHandler.ctx.channel(), nonPartitionedTopicName); + } + brokerService.getTopicIfExists(nonPartitionedTopicName).whenComplete((nonPartitionedTopic, ex) -> { + if (ex != null) { + handleGetTopicException(nonPartitionedTopicName, topicCompletableFuture, ex); + // Failed to getTopic from current broker, remove non-partitioned topic cache, + // which added in getTopicBroker. + removeTopicManagerCache(nonPartitionedTopicName); + return; + } + if (nonPartitionedTopic.isPresent()) { + PersistentTopic persistentTopic = (PersistentTopic) nonPartitionedTopic.get(); + topicCompletableFuture.complete(Optional.of(persistentTopic)); + } else { + log.error("[{}]Get empty non-partitioned topic for name {}", + requestHandler.ctx.channel(), nonPartitionedTopicName); + removeTopicManagerCache(nonPartitionedTopicName); + topicCompletableFuture.complete(Optional.empty()); + } + }); + return; } + log.error("[{}]Get empty topic for name {}", requestHandler.ctx.channel(), topicName); + removeTopicManagerCache(topicName); + topicCompletableFuture.complete(Optional.empty()); }); // cache for removing producer topics.put(topicName, topicCompletableFuture); return topicCompletableFuture; } + private void handleGetTopicException(@NonNull final String topicName, + @NonNull final CompletableFuture> topicCompletableFuture, + @NonNull final Throwable ex) { + // The ServiceUnitNotReadyException is retryable, so we should print a warning log instead of error log + if (ex instanceof BrokerServiceException.ServiceUnitNotReadyException) { + log.warn("[{}] Failed to getTopic {}: {}", + requestHandler.ctx.channel(), topicName, ex.getMessage()); + topicCompletableFuture.complete(Optional.empty()); + } else { + log.error("[{}] Failed to getTopic {}. exception:", + requestHandler.ctx.channel(), topicName, ex); + topicCompletableFuture.completeExceptionally(ex); + } + + } + public void registerProducerInPersistentTopic(String topicName, PersistentTopic persistentTopic) { if (closed.get()) { if (log.isDebugEnabled()) { diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/DistributedClusterTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/DistributedClusterTest.java index c67b963b55..3285bcaddd 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/DistributedClusterTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/DistributedClusterTest.java @@ -25,6 +25,7 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.time.Duration; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -36,6 +37,7 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.junit.Assert; import org.slf4j.Logger; @@ -432,6 +434,156 @@ public void testMultiBrokerUnloadReload() throws Exception { kConsumer2.close(); } + + /** + * Unit test for test in distributed cluster env produce and consume non-partitioned topic, + * verify it works well. + */ + @Test(timeOut = 30000) + public void testMultiBrokerProduceAndConsumeNonPartitionedTopic() throws Exception { + String kafkaTopicName = "kopMultiBrokerNonPartitionedTopic"; + String pulsarTopicName = "persistent://public/default/" + kafkaTopicName; + String kopNamespace = "public/default"; + int totalMsgs = 50; + List topicPartitions = + Collections.singletonList(new TopicPartition(pulsarTopicName, 0)); + String messageStrPrefix = "Message_" + kafkaTopicName + "_"; + KProducer kProducer1 = null; + KProducer kProducer2 = null; + KConsumer kConsumer1 = null; + KConsumer kConsumer2 = null; + // 0. Preparing: create non-partitioned topic. + pulsarService1.getAdminClient().topics().createNonPartitionedTopic(kafkaTopicName); + PartitionedTopicMetadata partitionedTopicMetadata = admin.topics().getPartitionedTopicMetadata(pulsarTopicName); + assertEquals(partitionedTopicMetadata.partitions, 0); + try { + // 1. check lookup result, use other kafka broker port as bootstrap port. + String result = admin.lookups().lookupTopic(pulsarTopicName); + log.info("Server address:{}", result); + int kafkaBrokerPort; + if (result.endsWith(String.valueOf(primaryBrokerPort))) { + kafkaBrokerPort = secondaryKafkaBrokerPort; + } else { + kafkaBrokerPort = primaryKafkaBrokerPort; + } + log.info("kafkaBrokerPort:{}", kafkaBrokerPort); + + // 2. produce consume message with Kafka producer. + kProducer1 = new KProducer(kafkaTopicName, false, kafkaBrokerPort, true); + kafkaPublishMessage(kProducer1, totalMsgs / 2, messageStrPrefix); + + kProducer2 = new KProducer(kafkaTopicName, false, kafkaBrokerPort, true); + kafkaPublishMessage(kProducer2, totalMsgs / 2, messageStrPrefix); + + kConsumer1 = new KConsumer(kafkaTopicName, kafkaBrokerPort, "consumer-group-1"); + kConsumer2 = new KConsumer(kafkaTopicName, kafkaBrokerPort, "consumer-group-2"); + + log.info("Partition size: {}, will consume and commitOffset for 2 consumers", + topicPartitions.size()); + kafkaConsumeCommitMessage(kConsumer1, totalMsgs, messageStrPrefix, topicPartitions); + kafkaConsumeCommitMessage(kConsumer2, totalMsgs, messageStrPrefix, topicPartitions); + + // 3. unload + log.info("Unload namespace, lookup will trigger another reload."); + pulsarService1.getAdminClient().namespaces().unload(kopNamespace); + + // 4. publish consume again + log.info("Re Publish / Consume again."); + kafkaPublishMessage(kProducer1, totalMsgs, messageStrPrefix); + kafkaConsumeCommitMessage(kConsumer1, totalMsgs, messageStrPrefix, topicPartitions); + kafkaConsumeCommitMessage(kConsumer2, totalMsgs, messageStrPrefix, topicPartitions); + + } finally { + if (kProducer1 != null) { + kProducer1.close(); + } + if (kProducer2 != null) { + kProducer2.close(); + } + if (kConsumer1 != null) { + kConsumer1.close(); + } + if (kConsumer2 != null) { + kConsumer2.close(); + } + pulsarService1.getAdminClient().topics().delete(pulsarTopicName); + } + + } + + /** + * Unit test for test in distributed cluster env produce and consume one partition topic, + * verify it works well (This test case is used to compare with non-partitioned topic behavior). + */ + @Test(timeOut = 30000) + public void testMultiBrokerProduceAndConsumeOnePartitionedTopic() throws Exception { + String kafkaTopicName = "kopMultiBrokerOnePartitionedTopic"; + String pulsarTopicName = "persistent://public/default/" + kafkaTopicName; + String kopNamespace = "public/default"; + int totalMsgs = 50; + List topicPartitions = + Collections.singletonList(new TopicPartition(kafkaTopicName, 0)); + String messageStrPrefix = "Message_" + kafkaTopicName + "_"; + KProducer kProducer1 = null; + KProducer kProducer2 = null; + KConsumer kConsumer1 = null; + KConsumer kConsumer2 = null; + // 0. Preparing: create non-partitioned topic. + pulsarService1.getAdminClient().topics().createPartitionedTopic(pulsarTopicName, 1); + try { + // 1. check lookup result. + String result = admin.lookups().lookupTopic(pulsarTopicName); + log.info("Server address:{}", result); + int kafkaPort; + if (result.endsWith(String.valueOf(primaryBrokerPort))) { + kafkaPort = secondaryKafkaBrokerPort; + } else { + kafkaPort = primaryKafkaBrokerPort; + } + + // 2. produce consume message with Kafka producer. + kProducer1 = new KProducer(kafkaTopicName, false, kafkaPort, true); + kafkaPublishMessage(kProducer1, totalMsgs / 2, messageStrPrefix); + + kProducer2 = new KProducer(kafkaTopicName, false, kafkaPort, true); + kafkaPublishMessage(kProducer2, totalMsgs / 2, messageStrPrefix); + + kConsumer1 = new KConsumer(kafkaTopicName, kafkaPort, "consumer-group-1"); + kConsumer2 = new KConsumer(kafkaTopicName, kafkaPort, "consumer-group-2"); + + log.info("Partition size: {}, will consume and commitOffset for 2 consumers", + topicPartitions.size()); + kafkaConsumeCommitMessage(kConsumer1, totalMsgs, messageStrPrefix, topicPartitions); + kafkaConsumeCommitMessage(kConsumer2, totalMsgs, messageStrPrefix, topicPartitions); + + // 3. unload + log.info("Unload namespace, lookup will trigger another reload."); + pulsarService1.getAdminClient().namespaces().unload(kopNamespace); + + // 4. publish consume again + log.info("Re Publish / Consume again."); + kafkaPublishMessage(kProducer1, totalMsgs, messageStrPrefix); + kafkaConsumeCommitMessage(kConsumer1, totalMsgs, messageStrPrefix, topicPartitions); + kafkaConsumeCommitMessage(kConsumer2, totalMsgs, messageStrPrefix, topicPartitions); + + } finally { + if (kProducer1 != null) { + kProducer1.close(); + } + if (kProducer2 != null) { + kProducer2.close(); + } + if (kConsumer1 != null) { + kConsumer1.close(); + } + if (kConsumer2 != null) { + kConsumer2.close(); + } + pulsarService1.getAdminClient().topics().deletePartitionedTopic(pulsarTopicName); + } + + } + @Test(timeOut = 30000) public void testOneBrokerShutdown() throws Exception { int partitionNumber = 10; diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaNonPartitionedTopicTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaNonPartitionedTopicTest.java new file mode 100644 index 0000000000..c7fdaf21a0 --- /dev/null +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaNonPartitionedTopicTest.java @@ -0,0 +1,108 @@ +/** + * 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 static org.testng.Assert.assertTrue; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import lombok.Cleanup; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.PartitionInfo; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Test unit for non-partitioned topic. + */ +public class KafkaNonPartitionedTopicTest extends KopProtocolHandlerTestBase { + + private static final String TENANT = "KafkaNonPartitionedTopicTest"; + private static final String NAMESPACE = "ns1"; + + + @BeforeClass + @Override + protected void setup() throws Exception { + super.resetConfig(); + + conf.setKafkaTenant(TENANT); + conf.setKafkaNamespace(NAMESPACE); + conf.setKafkaMetadataTenant("internal"); + conf.setKafkaMetadataNamespace("__kafka"); + conf.setEnableTransactionCoordinator(true); + + conf.setClusterName(super.configClusterName); + super.internalSetup(); + } + + @AfterClass + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Test(timeOut = 30000) + public void testNonPartitionedTopic() throws PulsarAdminException { + String topic = "persistent://" + TENANT + "/" + NAMESPACE + "/" + "testNonPartitionedTopic"; + admin.topics().createNonPartitionedTopic(topic); + try { + @Cleanup + KProducer kProducer = new KProducer(topic, false, getKafkaBrokerPort()); + + int totalMsgs = 50; + String messageStrPrefix = topic + "_message_"; + + for (int i = 0; i < totalMsgs; i++) { + String messageStr = messageStrPrefix + i; + kProducer.getProducer().send(new ProducerRecord<>(topic, i, messageStr)); + } + @Cleanup + KConsumer kConsumer = new KConsumer(topic, getKafkaBrokerPort(), "DemoKafkaOnPulsarConsumer"); + + kConsumer.getConsumer().subscribe(Collections.singleton(topic)); + + int i = 0; + while (i < totalMsgs) { + ConsumerRecords records = kConsumer.getConsumer().poll(Duration.ofSeconds(1)); + for (ConsumerRecord record : records) { + Integer key = record.key(); + assertEquals(messageStrPrefix + key.toString(), record.value()); + i++; + } + } + assertEquals(i, totalMsgs); + + // No more records + ConsumerRecords records = kConsumer.getConsumer().poll(Duration.ofMillis(200)); + assertTrue(records.isEmpty()); + + // Ensure that we can list the topic + Map> result = kConsumer + .getConsumer().listTopics(Duration.ofSeconds(1)); + assertEquals(result.size(), 1); + } finally { + admin.topics().delete(topic); + } + } + +} diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java index 6eb0d15f30..9b374f0616 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java @@ -708,7 +708,6 @@ public void testMetadataForNonPartitionedTopic(short version) throws Exception { responseFuture); final MetadataResponse response = (MetadataResponse) responseFuture.get(); assertEquals(response.topicMetadata().size(), 1); - assertEquals(response.errors().size(), 1); - assertEquals(response.errors().get(topic), Errors.INVALID_TOPIC_EXCEPTION); + assertEquals(response.errors().size(), 0); } }