From 59df84255200eea2f3f5807fb2eb96ada345d3f4 Mon Sep 17 00:00:00 2001 From: congbo Date: Wed, 16 Feb 2022 23:05:10 +0800 Subject: [PATCH 1/3] [Transaction] Fix individual ack batch message with transaction abort will redeliver Duplicate bit sit --- .../mledger/util/PositionAckSetUtil.java | 12 +++- .../service/AbstractBaseDispatcher.java | 24 +++++++- .../persistent/PersistentSubscription.java | 3 + .../pendingack/PendingAckHandle.java | 11 ++++ .../impl/PendingAckHandleDisabled.java | 5 ++ .../pendingack/impl/PendingAckHandleImpl.java | 11 ++++ .../client/impl/TransactionEndToEndTest.java | 61 ++++++++++++++++++- 7 files changed, 120 insertions(+), 7 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java index 47d4bc2eea057..166dbc7247118 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java @@ -46,12 +46,18 @@ public static void andAckSet(PositionImpl currentPosition, PositionImpl otherPos if (currentPosition == null || otherPosition == null) { return; } - BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet()); - BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(otherPosition.getAckSet()); + currentPosition.setAckSet(andAckSet(currentPosition.getAckSet(), otherPosition.getAckSet())); + } + + //This method is do `or` operation for ack set + public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) { + BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(firstAckSet); + BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(secondAckSet); thisAckSet.and(otherAckSet); - currentPosition.setAckSet(thisAckSet.toLongArray()); + long[] ackSet = thisAckSet.toLongArray(); thisAckSet.recycle(); otherAckSet.recycle(); + return ackSet; } //This method is compare two position which position is bigger than another one. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java index ec51fecccfe00..17d68f78efc99 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.service; +import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet; import com.google.common.collect.ImmutableList; import io.netty.buffer.ByteBuf; import java.util.ArrayList; @@ -36,6 +37,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.intercept.BrokerInterceptor; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter; +import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.service.plugin.EntryFilter; import org.apache.pulsar.broker.service.plugin.EntryFilterWithClassLoader; @@ -203,8 +205,28 @@ && trackDelayedDelivery(entry.getLedgerId(), entry.getEntryId(), msgMetadata)) { batchSizes.setBatchSize(i, batchSize); long[] ackSet = null; if (indexesAcks != null && cursor != null) { + PositionImpl position = PositionImpl.get(entry.getLedgerId(), entry.getEntryId()); ackSet = cursor - .getDeletedBatchIndexesAsLongArray(PositionImpl.get(entry.getLedgerId(), entry.getEntryId())); + .getDeletedBatchIndexesAsLongArray(position); + + // some batch messages ack bit sit will be in pendingAck state, so don't send all bit sit to consumer + if (subscription instanceof PersistentSubscription) { + PositionImpl positionInPendingAck = + ((PersistentSubscription) subscription).getPositionInPendingAck(position); + // if this position not in pendingAck state, don't need to do any op + if (positionInPendingAck != null) { + + if (positionInPendingAck.hasAckSet()) { + // need to or ackSet in pendingAck state and cursor ackSet which bit sit has been acked + if (ackSet != null) { + ackSet = andAckSet(ackSet, positionInPendingAck.getAckSet()); + } else { + // if actSet is null, use pendingAck ackSet + ackSet = positionInPendingAck.getAckSet(); + } + } + } + } if (ackSet != null) { indexesAcks.setIndexesAcks(i, Pair.of(batchSize, ackSet)); } else { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 6d74b531493f8..8f36bd56214fc 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -1084,6 +1084,9 @@ public Map getSubscriptionProperties() { return subscriptionProperties; } + public PositionImpl getPositionInPendingAck(PositionImpl position) { + return pendingAckHandle.getPositionInPendingAck(position); + } /** * Return a merged map that contains the cursor properties specified by used * (eg. when using compaction subscription) and the subscription properties. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java index 620db5c4b4814..65e50c866df78 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java @@ -157,4 +157,15 @@ CompletableFuture individualAcknowledgeMessage(TxnID txnID, List + * If it does not return null, it means this Position is in pendingAck and if it is batch Position, + * it will return the corresponding ackSet in pendingAck + * + * @param position {@link Position} witch need to get in pendingAck + * @return {@link Position} return the position in pendingAck + */ + PositionImpl getPositionInPendingAck(PositionImpl position); } \ No newline at end of file diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleDisabled.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleDisabled.java index 119072700dfaa..758c69b17b795 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleDisabled.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleDisabled.java @@ -100,4 +100,9 @@ public CompletableFuture close() { public boolean checkIfPendingAckStoreInit() { return false; } + + @Override + public PositionImpl getPositionInPendingAck(PositionImpl position) { + return null; + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java index f001b39972e13..ada278c88fc17 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java @@ -937,6 +937,17 @@ public boolean checkIfPendingAckStoreInit() { return this.pendingAckStoreFuture != null && this.pendingAckStoreFuture.isDone(); } + @Override + public PositionImpl getPositionInPendingAck(PositionImpl position) { + if (individualAckPositions != null) { + MutablePair positionPair = this.individualAckPositions.get(position); + if (positionPair != null) { + return positionPair.getLeft(); + } + } + return null; + } + protected void handleCacheRequest() { while (true) { Runnable runnable = acceptQueue.poll(); 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 bfa19b2496abb..330915f9f4e13 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 @@ -33,10 +33,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; - import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; - import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; @@ -104,6 +102,63 @@ public Object[][] enableBatch() { return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; } + @Test + private void testIndividualAckAbortFilterAckSetInPendingAckState() throws Exception { + final String topicName = NAMESPACE1 + "/testIndividualAckAbortFilterAckSetInPendingAckState"; + final int count = 9; + Producer producer = pulsarClient + .newProducer(Schema.INT32) + .topic(topicName) + .enableBatching(true) + .batchingMaxPublishDelay(1, TimeUnit.HOURS) + .batchingMaxMessages(count).create(); + + @Cleanup + Consumer consumer = pulsarClient + .newConsumer(Schema.INT32) + .topic(topicName) + .isAckReceiptEnabled(true) + .subscriptionName("test") + .subscriptionType(SubscriptionType.Shared) + .enableBatchIndexAcknowledgment(true) + .subscribe(); + + for (int i = 0; i < count; i++) { + producer.sendAsync(i); + } + + Transaction firstTransaction = getTxn(); + + Transaction secondTransaction = getTxn(); + + // firstTransaction ack the first three messages and don't end the firstTransaction + for (int i = 0; i < count / 3; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), firstTransaction).get(); + } + + // if secondTransaction abort we only can receive the middle three messages + for (int i = 0; i < count / 3; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), secondTransaction).get(); + } + + // consumer normal ack the last three messages + for (int i = 0; i < count / 3; i++) { + consumer.acknowledgeAsync(consumer.receive()).get(); + } + + // if secondTransaction abort we only can receive the middle three messages + secondTransaction.abort().get(); + + // can receive 3 4 5 bit sit message + for (int i = 0; i < count / 3; i++) { + assertEquals(consumer.receive().getValue().intValue(), i + 3); + } + + // can't receive message anymore + assertNull(consumer.receive(2, TimeUnit.SECONDS)); + + } + @Test(dataProvider="enableBatch") private void produceCommitTest(boolean enableBatch) throws Exception { @Cleanup @@ -630,7 +685,7 @@ private void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, Subscri } } - private Transaction getTxn() throws Exception { + public Transaction getTxn() throws Exception { return pulsarClient .newTransaction() .withTransactionTimeout(10, TimeUnit.SECONDS) From 5a758178f178e6707d1452ae7214bc41be398a10 Mon Sep 17 00:00:00 2001 From: congbo Date: Sat, 2 Apr 2022 09:46:35 +0800 Subject: [PATCH 2/3] Fix some comment --- .../org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java index 7e27c748b8108..2ba9f3fee133c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java @@ -48,7 +48,7 @@ public static void andAckSet(PositionImpl currentPosition, PositionImpl otherPos currentPosition.setAckSet(andAckSet(currentPosition.getAckSet(), otherPosition.getAckSet())); } - //This method is do `or` operation for ack set + //This method is do `and` operation for ack set public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) { BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(firstAckSet); BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(secondAckSet); From e0c61b98b227efb9a1a33e77733561ff6bcda174 Mon Sep 17 00:00:00 2001 From: congbobo184 Date: Thu, 16 Feb 2023 21:54:33 +0800 Subject: [PATCH 3/3] fix code style --- .../apache/pulsar/broker/service/AbstractBaseDispatcher.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java index 0e9e47c70a3fc..ef2fd80302a98 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java @@ -224,8 +224,9 @@ public int filterEntriesForConsumer(@Nullable MessageMetadata[] metadataArray, i ackSet = cursor .getDeletedBatchIndexesAsLongArray(position); // some batch messages ack bit sit will be in pendingAck state, so don't send all bit sit to consumer - if (subscription instanceof PersistentSubscription && - ((PersistentSubscription) subscription).getPendingAckHandle() instanceof PendingAckHandleImpl) { + if (subscription instanceof PersistentSubscription + && ((PersistentSubscription) subscription) + .getPendingAckHandle() instanceof PendingAckHandleImpl) { PositionImpl positionInPendingAck = ((PersistentSubscription) subscription).getPositionInPendingAck(position); // if this position not in pendingAck state, don't need to do any op