fix the expired PersistentTopic has a closed managedLedger that will throw a ManagedLedgerException (#862)

Fixes #861 

### Motivation
See #861 for the reason for the investigation.

Since the namespacebundle has a listener, it can be guaranteed that the topic corresponding to tcm can be removed from the cache after being unloaded and loaded. 

But for topic, it does not seem reasonable to create a listener for close, because for the server, there will be a large number of topics . 

One more thing, pulsar currently does not have a listener for the close method of a single topic. And there is no need to implement such a listener.

Author: wenbingshen

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

@@ -221,6 +221,15 @@ private CompletableFuture<Pair<ManagedCursor, Long>> asyncGetCursorByOffset(long
             return CompletableFuture.completedFuture(null);
         }
         final ManagedLedger ledger = topic.getManagedLedger();
+
+        if (((ManagedLedgerImpl) ledger).getState() == ManagedLedgerImpl.State.Closed) {
+            log.error("[{}] Async get cursor for offset {} failed, because current managedLedger has been closed",
+                    requestHandler.ctx.channel(), offset);
+            CompletableFuture<Pair<ManagedCursor, Long>> future = new CompletableFuture<>();
+            future.completeExceptionally(new Exception("Current managedLedger has been closed."));
+            return future;
+        }
+
         return ledger.asyncFindPosition(new OffsetSearchPredicate(offset)).thenApply(position -> {
             final String cursorName = "kop-consumer-cursor-" + topic.getName()
                     + "-" + position.getLedgerId() + "-" + position.getEntryId()

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

@@ -290,6 +290,36 @@ public void handleFetch() {
         });
     }
 
+    private void registerPrepareMetadataFailedEvent(long startPrepareMetadataNanos) {
+        statsLogger.getPrepareMetadataStats().registerFailedEvent(
+                MathUtils.elapsedNanos(startPrepareMetadataNanos), TimeUnit.NANOSECONDS);
+    }
+
+    private boolean checkOffsetOutOfRange(KafkaTopicConsumerManager tcm,
+                                          long offset,
+                                          TopicPartition topicPartition,
+                                          long startPrepareMetadataNanos) {
+        // handle offset out-of-range exception
+        ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) tcm.getManagedLedger();
+        long logEndOffset = MessageIdUtils.getLogEndOffset(managedLedger);
+        // TODO: Offset out-of-range checks are still incomplete
+        // We only check the case of `offset > logEndOffset` and `offset < LogStartOffset`
+        // is currently not handled.
+        // Because we found that the operation of obtaining `LogStartOffset`
+        // requires reading from disk,
+        // and such a time-consuming operation is likely to harm the performance of FETCH request.
+        // More discussions please refer to https://github.com/streamnative/kop/pull/531
+        if (offset > logEndOffset) {
+            log.error("Received request for offset {} for partition {}, "
+                            + "but we only have entries less than {}.",
+                    offset, topicPartition, logEndOffset);
+            registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
+            addErrorPartitionResponse(topicPartition, Errors.OFFSET_OUT_OF_RANGE);
+            return true;
+        }
+        return false;
+    }
+
     private void handlePartitionData(final TopicPartition topicPartition,
                                      final FetchRequest.PartitionData partitionData,
                                      final String fullTopicName,
@@ -300,91 +330,73 @@ private void handlePartitionData(final TopicPartition topicPartition,
         // the future that is returned by getTopicConsumerManager is always completed normally
         topicManager.getTopicConsumerManager(fullTopicName).thenAccept(tcm -> {
             if (tcm == null) {
-                statsLogger.getPrepareMetadataStats().registerFailedEvent(
-                        MathUtils.elapsedNanos(startPrepareMetadataNanos), TimeUnit.NANOSECONDS);
+                registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
                 // remove null future cache
                 KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName);
                 addErrorPartitionResponse(topicPartition, Errors.NOT_LEADER_FOR_PARTITION);
-                return;
-            }
-
-            // handle offset out-of-range exception
-            ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) tcm.getManagedLedger();
-            long logEndOffset = MessageIdUtils.getLogEndOffset(managedLedger);
-            // TODO: Offset out-of-range checks are still incomplete
-            // We only check the case of `offset > logEndOffset` and `offset < LogStartOffset`
-            // is currently not handled.
-            // Because we found that the operation of obtaining `LogStartOffset`
-            // requires reading from disk,
-            // and such a time-consuming operation is likely to harm the performance of FETCH request.
-            // More discussions please refer to https://github.com/streamnative/kop/pull/531
-            if (offset > logEndOffset) {
-                log.error("Received request for offset {} for partition {}, "
-                                + "but we only have entries less than {}.",
-                        offset, topicPartition, logEndOffset);
-                addErrorPartitionResponse(topicPartition, Errors.OFFSET_OUT_OF_RANGE);
-                return;
-            }
-
-            if (log.isDebugEnabled()) {
-                log.debug("Fetch for {}: remove tcm to get cursor for fetch offset: {} .",
-                        topicPartition, offset);
-            }
-
-            final CompletableFuture<Pair<ManagedCursor, Long>> cursorFuture =
-                    tcm.removeCursorFuture(offset);
-            if (cursorFuture == null) {
-                // tcm is closed, just return a NONE error because the channel may be still active
-                log.warn("[{}] KafkaTopicConsumerManager is closed, remove TCM of {}",
-                        requestHandler.ctx, fullTopicName);
-                KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName);
-                addErrorPartitionResponse(topicPartition, Errors.NONE);
-                return;
-            }
-
-            // cursorFuture is never completed exceptionally because ManagedLedgerImpl#asyncFindPosition
-            // is never completed exceptionally.
-            cursorFuture.thenAccept(cursorLongPair -> {
-                if (cursorLongPair == null) {
-                    log.warn("KafkaTopicConsumerManager.remove({}) return null for topic {}. "
-                                    + "Fetch for topic return error.",
-                            offset, topicPartition);
-                    addErrorPartitionResponse(topicPartition, Errors.NOT_LEADER_FOR_PARTITION);
-                    return;
+            } else if (!checkOffsetOutOfRange(tcm, offset, topicPartition, startPrepareMetadataNanos)) {
+                if (log.isDebugEnabled()) {
+                    log.debug("Fetch for {}: remove tcm to get cursor for fetch offset: {} .",
+                            topicPartition, offset);
                 }
 
-                final ManagedCursor cursor = cursorLongPair.getLeft();
-                final AtomicLong cursorOffset = new AtomicLong(cursorLongPair.getRight());
-
-                statsLogger.getPrepareMetadataStats().registerSuccessfulEvent(
-                        MathUtils.elapsedNanos(startPrepareMetadataNanos), TimeUnit.NANOSECONDS);
-                long adjustedMaxBytes = Math.min(partitionData.maxBytes, limitBytes.get());
-                readEntries(cursor, topicPartition, cursorOffset, adjustedMaxBytes)
-                .whenComplete((entries, throwable) -> {
-                    if (throwable != null) {
-                        tcm.deleteOneCursorAsync(cursorLongPair.getLeft(),
-                                "cursor.readEntry fail. deleteCursor");
-                        addErrorPartitionResponse(topicPartition, Errors.forException(throwable));
-                        return;
-                    }
-                    if (entries == null) {
-                        addErrorPartitionResponse(topicPartition,
-                                Errors.forException(new ApiException("Cursor is null")));
-                        return;
-                    }
-                    long readSize = entries.stream().mapToLong(Entry::getLength).sum();
-                    limitBytes.addAndGet(-1 * readSize);
-                    handleEntries(
-                            entries,
-                            topicPartition,
-                            partitionData,
-                            fullTopicName,
-                            tcm,
-                            cursor,
-                            cursorOffset,
-                            readCommitted);
-                });
-            });
+                final CompletableFuture<Pair<ManagedCursor, Long>> cursorFuture =
+                        tcm.removeCursorFuture(offset);
+                if (cursorFuture == null) {
+                    // tcm is closed, just return a NONE error because the channel may be still active
+                    log.warn("[{}] KafkaTopicConsumerManager is closed, remove TCM of {}",
+                            requestHandler.ctx, fullTopicName);
+                    registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
+                    KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName);
+                    addErrorPartitionResponse(topicPartition, Errors.NONE);
+                } else {
+                    cursorFuture.whenComplete((cursorLongPair, ex) -> {
+                        if (ex != null) {
+                            log.error("KafkaTopicConsumerManager.asyncGetCursorByOffset({}) failed for topic {}.",
+                                    offset, topicPartition, ex.getCause());
+                            registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
+                            KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName);
+                            addErrorPartitionResponse(topicPartition, Errors.NOT_LEADER_FOR_PARTITION);
+                        } else if (cursorLongPair == null) {
+                            log.warn("KafkaTopicConsumerManager.remove({}) return null for topic {}. "
+                                            + "Fetch for topic return error.",
+                                    offset, topicPartition);
+                            registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
+                            addErrorPartitionResponse(topicPartition, Errors.NOT_LEADER_FOR_PARTITION);
+                        } else {
+                            final ManagedCursor cursor = cursorLongPair.getLeft();
+                            final AtomicLong cursorOffset = new AtomicLong(cursorLongPair.getRight());
+
+                            statsLogger.getPrepareMetadataStats().registerSuccessfulEvent(
+                                    MathUtils.elapsedNanos(startPrepareMetadataNanos), TimeUnit.NANOSECONDS);
+                            long adjustedMaxBytes = Math.min(partitionData.maxBytes, limitBytes.get());
+                            readEntries(cursor, topicPartition, cursorOffset, adjustedMaxBytes)
+                                    .whenComplete((entries, throwable) -> {
+                                        if (throwable != null) {
+                                            tcm.deleteOneCursorAsync(cursorLongPair.getLeft(),
+                                                    "cursor.readEntry fail. deleteCursor");
+                                            addErrorPartitionResponse(topicPartition, Errors.forException(throwable));
+                                        } else if (entries == null) {
+                                            addErrorPartitionResponse(topicPartition,
+                                                    Errors.forException(new ApiException("Cursor is null")));
+                                        } else {
+                                            long readSize = entries.stream().mapToLong(Entry::getLength).sum();
+                                            limitBytes.addAndGet(-1 * readSize);
+                                            handleEntries(
+                                                    entries,
+                                                    topicPartition,
+                                                    partitionData,
+                                                    fullTopicName,
+                                                    tcm,
+                                                    cursor,
+                                                    cursorOffset,
+                                                    readCommitted);
+                                        }
+                                    });
+                        }
+                    });
+                }
+            }
         });
     }
 

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

@@ -22,6 +22,7 @@
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertSame;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
@@ -36,6 +37,7 @@
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -506,4 +508,38 @@ public void testTopicManagerClose() throws Exception {
         assertTrue(originalTcmList.get(0).isClosed());
         assertTrue(originalTcmList.get(1).isClosed());
     }
+
+    @Test(timeOut = 20000)
+    public void testUnloadTopic() throws Exception {
+        final String topic = "test-unload-topic";
+        final String fullTopicName = "persistent://public/default/" + topic + "-partition-0";
+        final int numPartitions = 1;
+        admin.topics().createPartitionedTopic(topic, numPartitions);
+
+        final int totalMessages = 5;
+        @Cleanup
+        final KafkaProducer<String, String> producer = new KafkaProducer<>(newKafkaProducerProperties());
+        int numMessages = 0;
+        while (numMessages < totalMessages) {
+            producer.send(new ProducerRecord<>(topic, null, "test-value" + numMessages)).get();
+            numMessages++;
+        }
+
+        // We first get KafkaTopicConsumerManager, and then unload topic,
+        // so that KafkaTopicConsumerManager will become invalid
+        CompletableFuture<KafkaTopicConsumerManager> tcm = kafkaTopicManager.getTopicConsumerManager(fullTopicName);
+        KafkaTopicConsumerManager topicConsumerManager = tcm.get();
+        // unload topic
+        admin.topics().unload(fullTopicName);
+
+        // This proves that ManagedLedger has been closed
+        // and that the newly added code has taken effect.
+        try {
+            topicConsumerManager.removeCursorFuture(totalMessages - 1).get();
+            fail("should have failed");
+        } catch (ExecutionException ex) {
+            assertTrue(ex.getCause().getMessage().contains("Current managedLedger has been closed."));
+        }
+
+    }
 }