[fix] [txn] do not take snapshot when no txn.

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

@@ -219,20 +219,11 @@ 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);
+                } else {
+                    completableFuture.completeExceptionally(new BrokerServiceException
+                            .ServiceUnitNotReadyException("TransactionBuffer recover failed"));
                 }
             }).exceptionally(exception -> {
                 log.error("Topic {}: TransactionBuffer recover failed", this.topic.getName(), exception.getCause());
@@ -268,6 +259,10 @@ public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceI
                     + "Please use a new transaction to send message."));
             return completableFuture;
         }
+        if (checkIfNoSnapshot()) {
+            this.snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(maxReadPosition)
+                    .thenRun(() -> changeToReadyStateFromNoSnapshot()).join();
+        }
         topic.getManagedLedger().asyncAddEntry(buffer, new AsyncCallbacks.AddEntryCallback() {
             @Override
             public void addComplete(Position position, ByteBuf entryData, Object ctx) {

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.