Fix pulsar client transaction test

huangdx0726 · Nov 13, 2020 · a662470 · a662470
1 parent 8f9861a
commit a662470
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 2 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/transaction/EndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/transaction/EndToEndTest.java
@@ -224,6 +224,7 @@ private void txnAckTest(boolean batchEnable, int maxBatchSize,
                 .subscriptionName("test")
                 .enableBatchIndexAcknowledgment(true)
                 .subscriptionType(subscriptionType)
+                .acknowledgmentGroupTime(0, TimeUnit.MICROSECONDS)
                 .subscribe();
 
         @Cleanup

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1156,7 +1156,8 @@ void messageReceived(MessageIdData messageId, int redeliveryCount, List<Long> ac
                     msgMetadata.recycle();
                     return;
                 }
-                msgId = new BatchMessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), getPartitionIndex(), messageId.getBatchIndex(), -1, BatchMessageAckerDisabled.INSTANCE);
+                BatchMessageAcker batchMessageAcker = BatchMessageAcker.newAcker(ackBitSet);
+                msgId = new BatchMessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), getPartitionIndex(), messageId.getBatchIndex(), -1, batchMessageAcker);
             }
 
             final MessageImpl<T> message = new MessageImpl<>(topicName.toString(), msgId, msgMetadata,

diff --git a/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -181,7 +181,9 @@ public void addAcknowledgment(MessageIdImpl msgId, AckType ackType, Map<String,
     public void addBatchIndexAcknowledgment(BatchMessageIdImpl msgId, int batchIndex, int batchSize, AckType ackType,
                                             Map<String, Long> properties, TransactionImpl txn) {
         if (acknowledgementGroupTimeMicros == 0 || !properties.isEmpty()) {
-            doImmediateBatchIndexAck(msgId, batchIndex, batchSize, ackType, properties, txn.getTxnIdMostBits(), txn.getTxnIdLeastBits());
+            doImmediateBatchIndexAck(msgId, batchIndex, batchSize, ackType, properties,
+                    txn == null ? -1 : txn.getTxnIdMostBits(),
+                    txn == null ? -1 : txn.getTxnIdLeastBits());
         } else if (ackType == AckType.Cumulative) {
             BitSetRecyclable bitSet = BitSetRecyclable.create();
             bitSet.set(0, batchSize);