Fix no snapshot for normal message

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

@@ -219,19 +219,7 @@ public CompletableFuture<Void> checkIfTBRecoverCompletely(boolean isTxnEnabled)
         } else {
             CompletableFuture<Void> completableFuture = new CompletableFuture<>();
             transactionBufferFuture.thenRun(() -> {
-                if (checkIfNoSnapshot()) {
-                    snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(maxReadPosition).thenRun(() -> {
-                        if (changeToReadyStateFromNoSnapshot()) {
-                            timer.newTimeout(TopicTransactionBuffer.this,
-                                    takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
-                        }
-                        completableFuture.complete(null);
-                    }).exceptionally(exception -> {
-                        log.error("Topic {} failed to take snapshot", this.topic.getName());
-                        completableFuture.completeExceptionally(exception);
-                        return null;
-                    });
-                } else {
+                if (checkIfNoSnapshot() || checkIfReady()) {
                     completableFuture.complete(null);
                 }
             }).exceptionally(exception -> {
@@ -268,6 +256,9 @@ public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceI
                     + "Please use a new transaction to send message."));
             return completableFuture;
         }
+        if (checkIfNoSnapshot()) {
+            takeAbortedTxnSnapshot(this.maxReadPosition);
+        }
         topic.getManagedLedger().asyncAddEntry(buffer, new AsyncCallbacks.AddEntryCallback() {
             @Override
             public void addComplete(Position position, ByteBuf entryData, Object ctx) {
@@ -436,17 +427,26 @@ private void handleLowWaterMark(TxnID txnID, long lowWaterMark) {
         }
     }
 
+    private void takeAbortedTxnSnapshot(Position maxReadPosition) {
+        this.snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(this.maxReadPosition)
+                .thenRun(() -> {
+                    if (checkIfNoSnapshot()) {
+                        changeToReadyStateFromNoSnapshot();
+                    }
+                });
+    }
+
     private void takeSnapshotByChangeTimes() {
         if (changeMaxReadPositionCount.get() >= takeSnapshotIntervalNumber) {
             this.changeMaxReadPositionCount.set(0);
-            this.snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(this.maxReadPosition);
+            takeAbortedTxnSnapshot(this.maxReadPosition);
         }
     }
 
     private void takeSnapshotByTimeout() {
         if (changeMaxReadPositionCount.get() > 0) {
             this.changeMaxReadPositionCount.set(0);
-            this.snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(this.maxReadPosition);
+            takeAbortedTxnSnapshot(this.maxReadPosition);
         }
         this.timer.newTimeout(TopicTransactionBuffer.this,
                 takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TopicTransactionBufferTest.java

@@ -343,6 +343,48 @@ public void testGetLastMessageIdsWithOngoingTransactions() throws Exception {
         assertGetLastMessageId(consumer, expectedLastMessageID1);
     }
 
+    @Test
+    public void testNormalProductionNoSnapshot() throws Exception {
+        String topic = "persistent://" + NAMESPACE1 + "/testNormalProductionNoSnapshot";
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topic)
+                .create();
+
+        PersistentTopic topicRef = null;
+        for (int i = 0; i < pulsarServiceList.size(); i++) {
+            try {
+                topicRef = (PersistentTopic) pulsarServiceList.get(i)
+                        .getBrokerService().getTopic(topic, true).get().get();
+                break;
+            } catch (Exception e) {
+            }
+        }
+        TopicTransactionBuffer transactionBuffer = (TopicTransactionBuffer) topicRef.getTransactionBuffer();
+
+        // check the transaction buffer state is NoSnapshot
+        Assert.assertEquals(transactionBuffer.getState(), TopicTransactionBufferState.State.NoSnapshot);
+
+        // send 5 original messages.
+        for (int i = 0; i < 5; i++) {
+            producer.newMessage().send();
+        }
+
+        // check the transaction buffer state is NoSnapshot
+        Assert.assertEquals(transactionBuffer.getState(), TopicTransactionBufferState.State.NoSnapshot);
+
+        // create consumer to get last message id, trigger the snapshot
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(topic)
+                .subscriptionName("sub")
+                .subscribe();
+        consumer.getLastMessageIds();
+
+        // check the transaction buffer state is Ready
+        Assert.assertEquals(transactionBuffer.getState(), TopicTransactionBufferState.State.NoSnapshot);
+    }
+
     /**
      * produce 3 messages and then trigger a ledger switch,
      * then create a transaction and send a transactional message.