Revert some use less change

Demogorgon314 · Jun 21, 2023 · 85570d1 · 85570d1
1 parent b18239e
commit 85570d1
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 15 deletions.
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
@@ -143,21 +143,15 @@ public void close() {
     }
 
     public CompletableFuture<Optional<PersistentTopic>> getTopic(String topicName) {
-        try {
-            if (closed.get()) {
-                if (log.isDebugEnabled()) {
-                    log.debug("[{}] Return null for getTopic({}) since channel is closing",
-                            requestHandler.ctx.channel(), topicName);
-                }
-                return CompletableFuture.completedFuture(Optional.empty());
+        if (closed.get()) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Return null for getTopic({}) since channel is closing",
+                        requestHandler.ctx.channel(), topicName);
             }
-            CompletableFuture<Optional<PersistentTopic>> topicCompletableFuture =
-                    kafkaTopicLookupService.getTopic(topicName, requestHandler.ctx.channel());
-            return topicCompletableFuture;
-        } catch (Throwable error) {
-            log.error("Unhandled error here for {}", topicName, error);
-            return CompletableFuture.completedFuture(Optional.empty());
         }
+        CompletableFuture<Optional<PersistentTopic>> topicCompletableFuture =
+                kafkaTopicLookupService.getTopic(topicName, requestHandler.ctx.channel());
+        return topicCompletableFuture;
     }
 
     public void invalidateCacheForFencedManagerLedgerOnTopic(String fullTopicName) {

diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaMessageOrderTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaMessageOrderTestBase.java
@@ -121,7 +121,7 @@ public void testKafkaProduceMessageOrder(int batchSize) throws Exception {
             @Cleanup
             KafkaProducer<Integer, String> producer = new KafkaProducer<>(props);
 
-            int totalMsgs = 10;
+            int totalMsgs = batchSize * 2 + batchSize / 2;
             String messageStrPrefix = "Message_Kop_KafkaProducePulsarConsumeOrder_";
 
             List<CompletableFuture<RecordMetadata>> handles = new ArrayList<>();

diff --git a/tests/src/test/resources/log4j2.xml b/tests/src/test/resources/log4j2.xml
@@ -43,6 +43,6 @@
         <Logger name="io.streamnative" level="debug"/>
         <Logger name="org.apache.pulsar" level="info"/>
         <Logger name="org.apache.bookkeeper" level="info"/>
-        <Logger name="org.apache.kafka" level="info"/>
+        <Logger name="org.apache.kafka" level="debug"/>
     </Loggers>
 </Configuration>