diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java index 220b4327cceac..12d5ee42cc7c8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java @@ -472,12 +472,14 @@ public void run() { try { while (reader.hasMoreEvents()) { Message message = reader.readNext(); - TransactionBufferSnapshot transactionBufferSnapshot = message.getValue(); - if (topic.getName().equals(transactionBufferSnapshot.getTopicName())) { - callBack.handleSnapshot(transactionBufferSnapshot); - this.startReadCursorPosition = PositionImpl.get( - transactionBufferSnapshot.getMaxReadPositionLedgerId(), - transactionBufferSnapshot.getMaxReadPositionEntryId()); + if (topic.getName().equals(message.getKey())) { + TransactionBufferSnapshot transactionBufferSnapshot = message.getValue(); + if (transactionBufferSnapshot != null) { + callBack.handleSnapshot(transactionBufferSnapshot); + this.startReadCursorPosition = PositionImpl.get( + transactionBufferSnapshot.getMaxReadPositionLedgerId(), + transactionBufferSnapshot.getMaxReadPositionEntryId()); + } } } } catch (PulsarClientException pulsarClientException) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index f971b5a770cca..d5ae61c012e50 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -377,6 +377,23 @@ private void txnAckTest(boolean batchEnable, int maxBatchSize, } } + @Test + public void testAfterDeleteTopicOtherTopicCanRecover() throws Exception { + String topicOne = "persistent://" + NAMESPACE1 + "/topic-one"; + String topicTwo = "persistent://" + NAMESPACE1 + "/topic-two"; + String sub = "test"; + admin.topics().createNonPartitionedTopic(topicOne); + admin.topics().createSubscription(topicOne, "test", MessageId.earliest); + admin.topics().delete(topicOne); + + Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicTwo).create(); + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicTwo).subscriptionName(sub).subscribe(); + String content = "test"; + producer.send(content); + assertEquals(consumer.receive().getValue(), content); + } + @Test public void txnMessageAckTest() throws Exception { String topic = TOPIC_MESSAGE_ACK_TEST;