[Transaction] Fix topicTransactionBuffer handle null snapshot (#12758)

fix #12754 Now when delete topic, we will write a null value to Transaction buffer snapshot topic, other topic recover by this transaction buffer snapshot system topic, will produce NPE judge NPE logic (cherry picked from commit c90c89b)
apache · Nov 12, 2021 · 94db24a · 94db24a
1 parent 28e3d79
commit 94db24a
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 8 deletions.
diff --git a/...rc/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/...rc/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -532,14 +532,16 @@ public void run() {
                 try {
                     boolean hasSnapshot = false;
                     while (reader.hasMoreEvents()) {
-                        hasSnapshot = true;
                         Message<TransactionBufferSnapshot> 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) {
+                                hasSnapshot = true;
+                                callBack.handleSnapshot(transactionBufferSnapshot);
+                                this.startReadCursorPosition = PositionImpl.get(
+                                        transactionBufferSnapshot.getMaxReadPositionLedgerId(),
+                                        transactionBufferSnapshot.getMaxReadPositionEntryId());
+                            }
                         }
                     }
                     if (!hasSnapshot) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -295,4 +295,4 @@ public void testTakeSnapshotBeforeBuildTxnProducer() throws Exception {
             Assert.assertEquals(snapshot1.getMaxReadPositionEntryId(), 1);
         });
     }
-}
+}
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
@@ -376,6 +376,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<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicTwo).create();
+        Consumer<String> 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;