You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/10/14 07:55:32 UTC
[GitHub] [pulsar] congbobo184 opened a new pull request #8256: Transaction pendingack server implement
congbobo184 opened a new pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256
to solve #7981
## Motivation
- we need to support transaction pending-ack stat, and we need to support bacth ack.
- the transaction with ack, we need to return response, so we do it in the pr.
## implement
- client ack with transaction will carry only this ack bit set 1 1 1 1 1 0 1 1，the 6 point is this transaction ack bit set point.
- client will save ack with transaction request stat, and it will time out.
this PR also solve #8071 , we don't add the incorrect bundle data int the server own cache
```
org.apache.pulsar.broker.namespace.OwnershipCache;
private CompletableFuture<Optional<Map.Entry<NamespaceEphemeralData, Stat>>> resolveOwnership(String path) {
return ownershipReadOnlyCache.getWithStatAsync(path).thenApply(optionalOwnerDataWithStat -> {
if (optionalOwnerDataWithStat.isPresent()) {
Map.Entry<NamespaceEphemeralData, Stat> ownerDataWithStat = optionalOwnerDataWithStat.get();
Stat stat = ownerDataWithStat.getValue();
if (stat.getEphemeralOwner() == localZkCache.getZooKeeper().getSessionId()) {
LOG.info("Successfully reestablish ownership of {}", path);
OwnedBundle ownedBundle = new OwnedBundle(ServiceUnitZkUtils.suBundleFromPath(path, bundleFactory));
if (optionalOwnerDataWithStat.toString().equals(selfOwnerInfo.toString())) {
ownedBundlesCache.put(path, CompletableFuture.completedFuture(ownedBundle));
}
ownershipReadOnlyCache.invalidate(path);
namespaceService.onNamespaceBundleOwned(ownedBundle.getNamespaceBundle());
}
}
return optionalOwnerDataWithStat;
});
}
```
### Verifying this change
Add the tests for it
Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes
Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-708861417
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511714578
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
Review comment:
OK, it will completeException(NotAllowException).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715669
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -331,7 +332,8 @@ private PersistentSubscription createPersistentSubscription(String subscriptionN
if (subscriptionName.equals(Compactor.COMPACTION_SUBSCRIPTION)) {
return new CompactorSubscription(this, compactedTopic, subscriptionName, cursor);
} else {
- return new PersistentSubscription(this, subscriptionName, cursor, replicated);
+ return new PersistentSubscription(this, subscriptionName, cursor,
+ replicated, new PendingAckHandleImpl(topic, subscriptionName));
Review comment:
yes, now the handle will initialize in persistentSubscription
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui closed pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui closed pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-721452463
Close this PR since the CI tests always skipped.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-716472993
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511716160
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
Review comment:
yes, we only instantiate with the PersistentSubscription
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-717254396
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715029
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
Review comment:
it will completeException(NotAllowException). because of it will completeException(), the exception will fill message, so the String should be retain。
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ checkArgument(positions.get(0) instanceof PositionImpl);
Review comment:
it will completeException(NotAllowException).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715966
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
+ *
+ * @param txnId TransactionID of an ongoing transaction trying to sck message.
+ * @param positions {@link Position}(s) it try to ack.
+ * @param ackType {@link PulsarApi.CommandAck.AckType}.
+ * cumulative ack or try to single ack message already acked by any ongoing transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> acknowledgeMessage(TxnID txnId, List<Position> positions, PulsarApi.CommandAck.AckType ackType);
+
+ /**
+ * Commit a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param properties Additional user-defined properties that can be associated with a particular cursor position.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> commitTxn(TxnID txnId, Map<String,Long> properties);
+
+ /**
+ * Abort a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param consumer {@link Consumer} which aborting transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer);
+
+ void setPersistentSubscription(PersistentSubscription persistentSubscription);
Review comment:
it has been delete and it will initialize by construction method
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513345226
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
Review comment:
1.
> The ack should not conflict if the cumulative ack position is greater than the last acked
```
if (i % 3 == 0) {
consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
}
```
it will check.
2.
> The ack should conflict if the cumulative ack position equals to the last acked
```
try {
consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
fail("not ack conflict ");
} catch (Exception e) {
Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
}
```
the message.getMessageId is equals the last acked.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715307
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
##########
@@ -81,6 +82,64 @@ public long getEntryId() {
return entryId;
}
+ public int compareToWithAckSet(PositionImpl other) {
+ checkNotNull(other);
+ int result = ComparisonChain.start().compare(this.ledgerId, other.ledgerId).compare(this.entryId, other.entryId)
+ .result();
+ if (result == 0) {
+ if (other.ackSet == null && ackSet == null) {
+ return result;
+ }
+ checkNotNull(other.ackSet);
+ checkNotNull(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ if (otherAckSet.equals(thisAckSet)) {
+ return result;
+ }
+ otherAckSet.and(thisAckSet);
+ boolean flag = otherAckSet.equals(thisAckSet);
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return flag ? 1 : 0;
+ }
+ return result;
+ }
+
+ public boolean isAckSetRepeated(PositionImpl other) {
+ if (ackSet == null || other.ackSet == null) {
+ return false;
+ }
+ checkNotNull(other);
+ checkNotNull(ackSet);
+ checkNotNull(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ if (otherAckSet.size() < thisAckSet.size()) {
+ otherAckSet.set(otherAckSet.size(), thisAckSet.size());
+ }
+ thisAckSet.flip(0, thisAckSet.size());
+ otherAckSet.flip(0, otherAckSet.size());
+ thisAckSet.and(otherAckSet);
+ boolean isAckSetRepeated = !thisAckSet.isEmpty();
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return isAckSetRepeated;
+ }
+
+ public void andAckSet(PositionImpl other) {
+ checkNotNull(other);
+ checkNotNull(ackSet);
+ checkNotNull(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ if (otherAckSet.size() < thisAckSet.size()) {
+ otherAckSet.set(otherAckSet.size(), thisAckSet.size());
+ }
+ thisAckSet.and(otherAckSet);
+ this.ackSet = thisAckSet.toLongArray();
+ }
Review comment:
it has move to PendingAckHandleImpl
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513366936
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ private Map<TxnID, HashMap<PositionImpl, PositionImpl>> pendingIndividualAckMessagesMap;
+
+ private Map<PositionImpl, PositionImpl> pendingAckMessages;
+
+ private Map<PositionImpl, HashSet<TxnID>> pendingAckBatchMessageMap;
+
+ private Pair<TxnID, PositionImpl> cumulativeAckPair;
+
+ private final String topicName;
+
+ private final String subName;
+
+ private final PersistentSubscription persistentSubscription;
+
+ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
+ this.topicName = persistentSubscription.getTopicName();
+ this.subName = persistentSubscription.getName();
+ this.persistentSubscription = persistentSubscription;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ if (txnId == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ }
+ if (positions == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
+ }
+ if (AckType.Cumulative == ackType) {
+ return acknowledgeMessageCumulative(positions, txnId);
+ } else {
+ return acknowledgeMessageIndividual(positions, txnId);
+ }
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageCumulative(List<Position> positions, TxnID txnID) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ if (!(positions.get(0) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ PositionImpl position = (PositionImpl) positions.get(0);
+
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnID.toString(), position);
+ }
+
+ if (this.cumulativeAckPair == null) {
+ this.cumulativeAckPair = MutablePair.of(txnID, position);
+ } else if (this.cumulativeAckPair.getKey().equals(txnID)
+ && compareToWithAckSetForCumulativeAck(position, this.cumulativeAckPair.getValue()) > 0) {
+ this.cumulativeAckPair.setValue(position);
+
+ } else {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative batch ack position: " + position + " within range of current " +
+ "currentPosition: " + this.cumulativeAckPair.getValue();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageIndividual(List<Position> positions, TxnID txnID) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ List<CompletableFuture<Void>> positionsFuture = new ArrayList<>(positions.size());
+ for (int i = 0; i < positions.size(); i++) {
+ if (!(positions.get(i) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid individual ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new NotAllowedException(errorMsg)));
+ continue;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Individual acks on {}", topicName, subName, txnID.toString(), positions);
+ }
+ if (pendingIndividualAckMessagesMap == null) {
+ pendingIndividualAckMessagesMap = new HashMap<>();
+ }
+
+ if (pendingAckMessages == null) {
+ pendingAckMessages = new HashMap<>();
+ }
+ PositionImpl position = (PositionImpl) positions.get(i);
+ // If try to ack message already acked by committed transaction or normal acknowledge, throw exception.
+ if (((ManagedCursorImpl) persistentSubscription.getCursor()).isMessageDeleted(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " already acked before.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ if (position.hasAckSet()) {
+
+ if (pendingAckMessages.containsKey(position)
+ && isAckSetRepeated(pendingAckMessages.get(position), position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack batch message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+
+ if (pendingAckBatchMessageMap == null) {
+ this.pendingAckBatchMessageMap = new HashMap<>();
+ }
+
+ HashSet<TxnID> txnSet = this.pendingAckBatchMessageMap
+ .computeIfAbsent(position, txn -> new HashSet<>());
+
+ if (pendingAckMessageForCurrentTxn.containsKey(position)) {
+ andAckSet(pendingAckMessageForCurrentTxn.get(position), position);
+ } else {
+ pendingAckMessageForCurrentTxn.put(position, position);
+ }
+
+ if (!pendingAckMessages.containsKey(position)) {
+ this.pendingAckMessages.put(position, position);
+ } else {
+ andAckSet(this.pendingAckMessages.get(position), position);
+ }
+ txnSet.add(txnID);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ } else {
+ if (this.pendingAckMessages.containsKey(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+ pendingAckMessageForCurrentTxn.put(position, position);
+ this.pendingAckMessages.put(position, position);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ }
+ }
+ FutureUtil.waitForAll(positionsFuture).whenComplete((v, e) -> {
+ if (e != null) {
+ completableFuture.completeExceptionally(e);
+ } else {
+ completableFuture.complete(null);
+ }
+ });
+ return completableFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> commitTxn(TxnID txnId, Map<String, Long> properties) {
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ // It's valid to create transaction then commit without doing any operation, which will cause
+ // pendingAckMessagesMap to be null.
+ if (this.cumulativeAckPair != null) {
+ if (cumulativeAckPair.getKey().equals(txnId)) {
+ persistentSubscription.acknowledgeMessage(Collections
+ .singletonList(this.cumulativeAckPair.getValue()), AckType.Cumulative, properties);
+ this.cumulativeAckPair = null;
+ }
+ } else {
+ if (pendingIndividualAckMessagesMap != null && pendingIndividualAckMessagesMap.containsKey(txnId)) {
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null
+ && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet = pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ persistentSubscription.acknowledgeMessage(new ArrayList<>(pendingAckMessageForCurrentTxn.values()),
+ AckType.Individual, properties);
+ }
+ }
+ }
+ commitFuture.complete(null);
+ return commitFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ if (this.cumulativeAckPair != null) {
+ if (this.cumulativeAckPair.getKey().equals(txnId)) {
+ this.cumulativeAckPair = null;
+ redeliverUnacknowledgedMessages(consumer);
+ }
+ } else if (this.pendingIndividualAckMessagesMap != null){
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet =
+ pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ this.pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ redeliverUnacknowledgedMessages(consumer, new ArrayList<>(pendingAckMessageForCurrentTxn.values()));
+ }
+ }
+ abortFuture.complete(null);
+ return abortFuture;
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer) {
+ ConcurrentLongLongPairHashMap positionMap = consumer.getPendingAcks();
+ // Only check if message is in pending_ack status when there's ongoing transaction.
+ if (null != positionMap && ((pendingAckMessages != null && pendingAckMessages.size() != 0)
+ || this.cumulativeAckPair != null)) {
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positionMap.asMap().forEach((key, value) -> {
+ PositionImpl position = new PositionImpl(key.first, key.second);
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition);
+ });
+
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer);
+ }
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
+ // If there's ongoing transaction.
+ if ((pendingAckMessages != null && pendingAckMessages.size() != 0) || this.cumulativeAckPair != null) {
+ // Check if message is in pending_ack status.
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positions.forEach(position ->
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition));
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, positions);
+ }
+ }
+
+ private void redeliverUnacknowledgedMessagesCommon(PositionImpl position, List<PositionImpl> pendingPositions,
+ PositionImpl cumulativeAckPosition) {
+ if (position.hasAckSet()) {
+ if (this.pendingAckMessages != null) {
+ if (this.pendingAckMessages.containsKey(position)) {
+ if (!isAckSetRepeated(this.pendingAckMessages.get(position), position)) {
+ pendingPositions.add(position);
+ }
+ } else {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null) {
+ int compareNumber = position.compareTo((cumulativeAckPosition));
+ if (compareNumber > 0 || (compareNumber == 0 && !isAckSetRepeated(position, cumulativeAckPosition))) {
+ pendingPositions.add(position);
+ }
+ }
+ } else {
+ if (this.pendingAckMessages != null) {
+ if (!this.pendingAckMessages.containsKey(position)) {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null ) {
+ if (position.compareTo(cumulativeAckPosition) > 0) {
+ pendingPositions.add(position);
+ }
+ }
+ }
+ }
+
+ public int compareToWithAckSetForCumulativeAck(PositionImpl currentPosition,PositionImpl otherPosition) {
+ int result = ComparisonChain.start().compare(currentPosition.getLedgerId(),
+ otherPosition.getLedgerId()).compare(currentPosition.getEntryId(), otherPosition.getEntryId())
+ .result();
+ if (result == 0) {
+ if (otherPosition.getAckSet() == null && currentPosition.getAckSet() == null) {
+ return result;
+ }
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(otherPosition.getAckSet());
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet());
+ if (otherAckSet.equals(thisAckSet)) {
+ return result;
+ }
+ otherAckSet.and(thisAckSet);
+ boolean flag = otherAckSet.equals(thisAckSet);
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return flag ? 1 : 0;
+ }
+ return result;
+ }
+
+ public boolean isAckSetRepeated(PositionImpl currentPosition, PositionImpl otherPosition) {
Review comment:
```suggestion
public boolean isAckSetOverlap(PositionImpl currentPosition, PositionImpl otherPosition) {
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ private Map<TxnID, HashMap<PositionImpl, PositionImpl>> pendingIndividualAckMessagesMap;
+
+ private Map<PositionImpl, PositionImpl> pendingAckMessages;
+
+ private Map<PositionImpl, HashSet<TxnID>> pendingAckBatchMessageMap;
+
+ private Pair<TxnID, PositionImpl> cumulativeAckPair;
+
+ private final String topicName;
+
+ private final String subName;
+
+ private final PersistentSubscription persistentSubscription;
+
+ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
+ this.topicName = persistentSubscription.getTopicName();
+ this.subName = persistentSubscription.getName();
+ this.persistentSubscription = persistentSubscription;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ if (txnId == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ }
+ if (positions == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
+ }
+ if (AckType.Cumulative == ackType) {
+ return acknowledgeMessageCumulative(positions, txnId);
+ } else {
+ return acknowledgeMessageIndividual(positions, txnId);
+ }
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageCumulative(List<Position> positions, TxnID txnID) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ if (!(positions.get(0) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ PositionImpl position = (PositionImpl) positions.get(0);
+
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnID.toString(), position);
+ }
+
+ if (this.cumulativeAckPair == null) {
+ this.cumulativeAckPair = MutablePair.of(txnID, position);
+ } else if (this.cumulativeAckPair.getKey().equals(txnID)
+ && compareToWithAckSetForCumulativeAck(position, this.cumulativeAckPair.getValue()) > 0) {
+ this.cumulativeAckPair.setValue(position);
+
+ } else {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative batch ack position: " + position + " within range of current " +
+ "currentPosition: " + this.cumulativeAckPair.getValue();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageIndividual(List<Position> positions, TxnID txnID) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ List<CompletableFuture<Void>> positionsFuture = new ArrayList<>(positions.size());
+ for (int i = 0; i < positions.size(); i++) {
+ if (!(positions.get(i) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid individual ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new NotAllowedException(errorMsg)));
+ continue;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Individual acks on {}", topicName, subName, txnID.toString(), positions);
+ }
+ if (pendingIndividualAckMessagesMap == null) {
+ pendingIndividualAckMessagesMap = new HashMap<>();
+ }
+
+ if (pendingAckMessages == null) {
+ pendingAckMessages = new HashMap<>();
+ }
+ PositionImpl position = (PositionImpl) positions.get(i);
+ // If try to ack message already acked by committed transaction or normal acknowledge, throw exception.
+ if (((ManagedCursorImpl) persistentSubscription.getCursor()).isMessageDeleted(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " already acked before.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ if (position.hasAckSet()) {
+
+ if (pendingAckMessages.containsKey(position)
+ && isAckSetRepeated(pendingAckMessages.get(position), position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack batch message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+
+ if (pendingAckBatchMessageMap == null) {
+ this.pendingAckBatchMessageMap = new HashMap<>();
+ }
+
+ HashSet<TxnID> txnSet = this.pendingAckBatchMessageMap
+ .computeIfAbsent(position, txn -> new HashSet<>());
+
+ if (pendingAckMessageForCurrentTxn.containsKey(position)) {
+ andAckSet(pendingAckMessageForCurrentTxn.get(position), position);
+ } else {
+ pendingAckMessageForCurrentTxn.put(position, position);
+ }
+
+ if (!pendingAckMessages.containsKey(position)) {
+ this.pendingAckMessages.put(position, position);
+ } else {
+ andAckSet(this.pendingAckMessages.get(position), position);
+ }
+ txnSet.add(txnID);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ } else {
+ if (this.pendingAckMessages.containsKey(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+ pendingAckMessageForCurrentTxn.put(position, position);
+ this.pendingAckMessages.put(position, position);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ }
+ }
+ FutureUtil.waitForAll(positionsFuture).whenComplete((v, e) -> {
+ if (e != null) {
+ completableFuture.completeExceptionally(e);
+ } else {
+ completableFuture.complete(null);
+ }
+ });
+ return completableFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> commitTxn(TxnID txnId, Map<String, Long> properties) {
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ // It's valid to create transaction then commit without doing any operation, which will cause
+ // pendingAckMessagesMap to be null.
+ if (this.cumulativeAckPair != null) {
+ if (cumulativeAckPair.getKey().equals(txnId)) {
+ persistentSubscription.acknowledgeMessage(Collections
+ .singletonList(this.cumulativeAckPair.getValue()), AckType.Cumulative, properties);
+ this.cumulativeAckPair = null;
+ }
+ } else {
+ if (pendingIndividualAckMessagesMap != null && pendingIndividualAckMessagesMap.containsKey(txnId)) {
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null
+ && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet = pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ persistentSubscription.acknowledgeMessage(new ArrayList<>(pendingAckMessageForCurrentTxn.values()),
+ AckType.Individual, properties);
+ }
+ }
+ }
+ commitFuture.complete(null);
+ return commitFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ if (this.cumulativeAckPair != null) {
+ if (this.cumulativeAckPair.getKey().equals(txnId)) {
+ this.cumulativeAckPair = null;
+ redeliverUnacknowledgedMessages(consumer);
+ }
+ } else if (this.pendingIndividualAckMessagesMap != null){
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet =
+ pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ this.pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ redeliverUnacknowledgedMessages(consumer, new ArrayList<>(pendingAckMessageForCurrentTxn.values()));
+ }
+ }
+ abortFuture.complete(null);
+ return abortFuture;
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer) {
+ ConcurrentLongLongPairHashMap positionMap = consumer.getPendingAcks();
+ // Only check if message is in pending_ack status when there's ongoing transaction.
+ if (null != positionMap && ((pendingAckMessages != null && pendingAckMessages.size() != 0)
+ || this.cumulativeAckPair != null)) {
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positionMap.asMap().forEach((key, value) -> {
+ PositionImpl position = new PositionImpl(key.first, key.second);
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition);
+ });
+
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer);
+ }
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
+ // If there's ongoing transaction.
+ if ((pendingAckMessages != null && pendingAckMessages.size() != 0) || this.cumulativeAckPair != null) {
+ // Check if message is in pending_ack status.
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positions.forEach(position ->
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition));
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, positions);
+ }
+ }
+
+ private void redeliverUnacknowledgedMessagesCommon(PositionImpl position, List<PositionImpl> pendingPositions,
+ PositionImpl cumulativeAckPosition) {
+ if (position.hasAckSet()) {
+ if (this.pendingAckMessages != null) {
+ if (this.pendingAckMessages.containsKey(position)) {
+ if (!isAckSetRepeated(this.pendingAckMessages.get(position), position)) {
+ pendingPositions.add(position);
+ }
+ } else {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null) {
+ int compareNumber = position.compareTo((cumulativeAckPosition));
+ if (compareNumber > 0 || (compareNumber == 0 && !isAckSetRepeated(position, cumulativeAckPosition))) {
+ pendingPositions.add(position);
+ }
+ }
+ } else {
+ if (this.pendingAckMessages != null) {
+ if (!this.pendingAckMessages.containsKey(position)) {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null ) {
+ if (position.compareTo(cumulativeAckPosition) > 0) {
+ pendingPositions.add(position);
+ }
+ }
+ }
+ }
+
+ public int compareToWithAckSetForCumulativeAck(PositionImpl currentPosition,PositionImpl otherPosition) {
+ int result = ComparisonChain.start().compare(currentPosition.getLedgerId(),
+ otherPosition.getLedgerId()).compare(currentPosition.getEntryId(), otherPosition.getEntryId())
+ .result();
+ if (result == 0) {
+ if (otherPosition.getAckSet() == null && currentPosition.getAckSet() == null) {
+ return result;
+ }
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(otherPosition.getAckSet());
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet());
+ if (otherAckSet.equals(thisAckSet)) {
+ return result;
+ }
+ otherAckSet.and(thisAckSet);
+ boolean flag = otherAckSet.equals(thisAckSet);
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return flag ? 1 : 0;
+ }
+ return result;
+ }
+
+ public boolean isAckSetRepeated(PositionImpl currentPosition, PositionImpl otherPosition) {
+ if (currentPosition.getAckSet() == null || otherPosition.getAckSet() == null) {
Review comment:
Add check to ensure the ledger Id and the entry Id of the current position equals to the other position
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ private Map<TxnID, HashMap<PositionImpl, PositionImpl>> pendingIndividualAckMessagesMap;
+
+ private Map<PositionImpl, PositionImpl> pendingAckMessages;
+
+ private Map<PositionImpl, HashSet<TxnID>> pendingAckBatchMessageMap;
+
+ private Pair<TxnID, PositionImpl> cumulativeAckPair;
+
+ private final String topicName;
+
+ private final String subName;
+
+ private final PersistentSubscription persistentSubscription;
+
+ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
+ this.topicName = persistentSubscription.getTopicName();
+ this.subName = persistentSubscription.getName();
+ this.persistentSubscription = persistentSubscription;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ if (txnId == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ }
+ if (positions == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
+ }
+ if (AckType.Cumulative == ackType) {
+ return acknowledgeMessageCumulative(positions, txnId);
+ } else {
+ return acknowledgeMessageIndividual(positions, txnId);
+ }
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageCumulative(List<Position> positions, TxnID txnID) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ if (!(positions.get(0) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ PositionImpl position = (PositionImpl) positions.get(0);
+
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnID.toString(), position);
+ }
+
+ if (this.cumulativeAckPair == null) {
+ this.cumulativeAckPair = MutablePair.of(txnID, position);
+ } else if (this.cumulativeAckPair.getKey().equals(txnID)
+ && compareToWithAckSetForCumulativeAck(position, this.cumulativeAckPair.getValue()) > 0) {
+ this.cumulativeAckPair.setValue(position);
+
+ } else {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative batch ack position: " + position + " within range of current " +
+ "currentPosition: " + this.cumulativeAckPair.getValue();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageIndividual(List<Position> positions, TxnID txnID) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ List<CompletableFuture<Void>> positionsFuture = new ArrayList<>(positions.size());
+ for (int i = 0; i < positions.size(); i++) {
+ if (!(positions.get(i) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid individual ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new NotAllowedException(errorMsg)));
+ continue;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Individual acks on {}", topicName, subName, txnID.toString(), positions);
+ }
+ if (pendingIndividualAckMessagesMap == null) {
+ pendingIndividualAckMessagesMap = new HashMap<>();
+ }
+
+ if (pendingAckMessages == null) {
+ pendingAckMessages = new HashMap<>();
+ }
+ PositionImpl position = (PositionImpl) positions.get(i);
+ // If try to ack message already acked by committed transaction or normal acknowledge, throw exception.
+ if (((ManagedCursorImpl) persistentSubscription.getCursor()).isMessageDeleted(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " already acked before.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ if (position.hasAckSet()) {
+
+ if (pendingAckMessages.containsKey(position)
+ && isAckSetRepeated(pendingAckMessages.get(position), position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack batch message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+
+ if (pendingAckBatchMessageMap == null) {
+ this.pendingAckBatchMessageMap = new HashMap<>();
+ }
+
+ HashSet<TxnID> txnSet = this.pendingAckBatchMessageMap
+ .computeIfAbsent(position, txn -> new HashSet<>());
+
+ if (pendingAckMessageForCurrentTxn.containsKey(position)) {
+ andAckSet(pendingAckMessageForCurrentTxn.get(position), position);
+ } else {
+ pendingAckMessageForCurrentTxn.put(position, position);
+ }
+
+ if (!pendingAckMessages.containsKey(position)) {
+ this.pendingAckMessages.put(position, position);
+ } else {
+ andAckSet(this.pendingAckMessages.get(position), position);
+ }
+ txnSet.add(txnID);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ } else {
+ if (this.pendingAckMessages.containsKey(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+ pendingAckMessageForCurrentTxn.put(position, position);
+ this.pendingAckMessages.put(position, position);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ }
+ }
+ FutureUtil.waitForAll(positionsFuture).whenComplete((v, e) -> {
+ if (e != null) {
+ completableFuture.completeExceptionally(e);
+ } else {
+ completableFuture.complete(null);
+ }
+ });
+ return completableFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> commitTxn(TxnID txnId, Map<String, Long> properties) {
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ // It's valid to create transaction then commit without doing any operation, which will cause
+ // pendingAckMessagesMap to be null.
+ if (this.cumulativeAckPair != null) {
+ if (cumulativeAckPair.getKey().equals(txnId)) {
+ persistentSubscription.acknowledgeMessage(Collections
+ .singletonList(this.cumulativeAckPair.getValue()), AckType.Cumulative, properties);
+ this.cumulativeAckPair = null;
+ }
+ } else {
+ if (pendingIndividualAckMessagesMap != null && pendingIndividualAckMessagesMap.containsKey(txnId)) {
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null
+ && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet = pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ persistentSubscription.acknowledgeMessage(new ArrayList<>(pendingAckMessageForCurrentTxn.values()),
+ AckType.Individual, properties);
+ }
+ }
+ }
+ commitFuture.complete(null);
+ return commitFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ if (this.cumulativeAckPair != null) {
+ if (this.cumulativeAckPair.getKey().equals(txnId)) {
+ this.cumulativeAckPair = null;
+ redeliverUnacknowledgedMessages(consumer);
+ }
+ } else if (this.pendingIndividualAckMessagesMap != null){
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet =
+ pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ this.pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ redeliverUnacknowledgedMessages(consumer, new ArrayList<>(pendingAckMessageForCurrentTxn.values()));
+ }
+ }
+ abortFuture.complete(null);
+ return abortFuture;
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer) {
+ ConcurrentLongLongPairHashMap positionMap = consumer.getPendingAcks();
+ // Only check if message is in pending_ack status when there's ongoing transaction.
+ if (null != positionMap && ((pendingAckMessages != null && pendingAckMessages.size() != 0)
+ || this.cumulativeAckPair != null)) {
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positionMap.asMap().forEach((key, value) -> {
+ PositionImpl position = new PositionImpl(key.first, key.second);
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition);
+ });
+
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer);
+ }
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
+ // If there's ongoing transaction.
+ if ((pendingAckMessages != null && pendingAckMessages.size() != 0) || this.cumulativeAckPair != null) {
+ // Check if message is in pending_ack status.
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positions.forEach(position ->
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition));
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, positions);
+ }
+ }
+
+ private void redeliverUnacknowledgedMessagesCommon(PositionImpl position, List<PositionImpl> pendingPositions,
+ PositionImpl cumulativeAckPosition) {
+ if (position.hasAckSet()) {
+ if (this.pendingAckMessages != null) {
+ if (this.pendingAckMessages.containsKey(position)) {
+ if (!isAckSetRepeated(this.pendingAckMessages.get(position), position)) {
+ pendingPositions.add(position);
+ }
+ } else {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null) {
+ int compareNumber = position.compareTo((cumulativeAckPosition));
+ if (compareNumber > 0 || (compareNumber == 0 && !isAckSetRepeated(position, cumulativeAckPosition))) {
+ pendingPositions.add(position);
+ }
+ }
+ } else {
+ if (this.pendingAckMessages != null) {
+ if (!this.pendingAckMessages.containsKey(position)) {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null ) {
+ if (position.compareTo(cumulativeAckPosition) > 0) {
+ pendingPositions.add(position);
+ }
+ }
+ }
+ }
+
+ public int compareToWithAckSetForCumulativeAck(PositionImpl currentPosition,PositionImpl otherPosition) {
+ int result = ComparisonChain.start().compare(currentPosition.getLedgerId(),
Review comment:
Check currentPosition and otherPosition not null.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ private Map<TxnID, HashMap<PositionImpl, PositionImpl>> pendingIndividualAckMessagesMap;
+
+ private Map<PositionImpl, PositionImpl> pendingAckMessages;
+
+ private Map<PositionImpl, HashSet<TxnID>> pendingAckBatchMessageMap;
+
+ private Pair<TxnID, PositionImpl> cumulativeAckPair;
+
+ private final String topicName;
+
+ private final String subName;
+
+ private final PersistentSubscription persistentSubscription;
+
+ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
+ this.topicName = persistentSubscription.getTopicName();
+ this.subName = persistentSubscription.getName();
+ this.persistentSubscription = persistentSubscription;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ if (txnId == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ }
+ if (positions == null) {
+ return FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
+ }
+ if (AckType.Cumulative == ackType) {
+ return acknowledgeMessageCumulative(positions, txnId);
+ } else {
+ return acknowledgeMessageIndividual(positions, txnId);
+ }
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageCumulative(List<Position> positions, TxnID txnID) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ if (!(positions.get(0) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid cumulative ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new NotAllowedException(errorMsg));
+ }
+
+ PositionImpl position = (PositionImpl) positions.get(0);
+
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnID.toString(), position);
+ }
+
+ if (this.cumulativeAckPair == null) {
+ this.cumulativeAckPair = MutablePair.of(txnID, position);
+ } else if (this.cumulativeAckPair.getKey().equals(txnID)
+ && compareToWithAckSetForCumulativeAck(position, this.cumulativeAckPair.getValue()) > 0) {
+ this.cumulativeAckPair.setValue(position);
+
+ } else {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to cumulative batch ack position: " + position + " within range of current " +
+ "currentPosition: " + this.cumulativeAckPair.getValue();
+ log.error(errorMsg);
+ return FutureUtil.failedFuture(new TransactionConflictException(errorMsg));
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private CompletableFuture<Void> acknowledgeMessageIndividual(List<Position> positions, TxnID txnID) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ List<CompletableFuture<Void>> positionsFuture = new ArrayList<>(positions.size());
+ for (int i = 0; i < positions.size(); i++) {
+ if (!(positions.get(i) instanceof PositionImpl)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " invalid individual ack received with position is not PositionImpl";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new NotAllowedException(errorMsg)));
+ continue;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Individual acks on {}", topicName, subName, txnID.toString(), positions);
+ }
+ if (pendingIndividualAckMessagesMap == null) {
+ pendingIndividualAckMessagesMap = new HashMap<>();
+ }
+
+ if (pendingAckMessages == null) {
+ pendingAckMessages = new HashMap<>();
+ }
+ PositionImpl position = (PositionImpl) positions.get(i);
+ // If try to ack message already acked by committed transaction or normal acknowledge, throw exception.
+ if (((ManagedCursorImpl) persistentSubscription.getCursor()).isMessageDeleted(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " already acked before.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ if (position.hasAckSet()) {
+
+ if (pendingAckMessages.containsKey(position)
+ && isAckSetRepeated(pendingAckMessages.get(position), position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack batch message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+
+ if (pendingAckBatchMessageMap == null) {
+ this.pendingAckBatchMessageMap = new HashMap<>();
+ }
+
+ HashSet<TxnID> txnSet = this.pendingAckBatchMessageMap
+ .computeIfAbsent(position, txn -> new HashSet<>());
+
+ if (pendingAckMessageForCurrentTxn.containsKey(position)) {
+ andAckSet(pendingAckMessageForCurrentTxn.get(position), position);
+ } else {
+ pendingAckMessageForCurrentTxn.put(position, position);
+ }
+
+ if (!pendingAckMessages.containsKey(position)) {
+ this.pendingAckMessages.put(position, position);
+ } else {
+ andAckSet(this.pendingAckMessages.get(position), position);
+ }
+ txnSet.add(txnID);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ } else {
+ if (this.pendingAckMessages.containsKey(position)) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnID +
+ " try to ack message:" + position + " in pending ack status.";
+ log.error(errorMsg);
+ positionsFuture.add(FutureUtil.failedFuture(new TransactionConflictException(errorMsg)));
+ continue;
+ }
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.computeIfAbsent(txnID, txn -> new HashMap<>());
+ pendingAckMessageForCurrentTxn.put(position, position);
+ this.pendingAckMessages.put(position, position);
+ positionsFuture.add(CompletableFuture.completedFuture(null));
+ }
+ }
+ FutureUtil.waitForAll(positionsFuture).whenComplete((v, e) -> {
+ if (e != null) {
+ completableFuture.completeExceptionally(e);
+ } else {
+ completableFuture.complete(null);
+ }
+ });
+ return completableFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> commitTxn(TxnID txnId, Map<String, Long> properties) {
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ // It's valid to create transaction then commit without doing any operation, which will cause
+ // pendingAckMessagesMap to be null.
+ if (this.cumulativeAckPair != null) {
+ if (cumulativeAckPair.getKey().equals(txnId)) {
+ persistentSubscription.acknowledgeMessage(Collections
+ .singletonList(this.cumulativeAckPair.getValue()), AckType.Cumulative, properties);
+ this.cumulativeAckPair = null;
+ }
+ } else {
+ if (pendingIndividualAckMessagesMap != null && pendingIndividualAckMessagesMap.containsKey(txnId)) {
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null
+ && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet = pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ persistentSubscription.acknowledgeMessage(new ArrayList<>(pendingAckMessageForCurrentTxn.values()),
+ AckType.Individual, properties);
+ }
+ }
+ }
+ commitFuture.complete(null);
+ return commitFuture;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ if (this.cumulativeAckPair != null) {
+ if (this.cumulativeAckPair.getKey().equals(txnId)) {
+ this.cumulativeAckPair = null;
+ redeliverUnacknowledgedMessages(consumer);
+ }
+ } else if (this.pendingIndividualAckMessagesMap != null){
+ HashMap<PositionImpl, PositionImpl> pendingAckMessageForCurrentTxn =
+ pendingIndividualAckMessagesMap.remove(txnId);
+ if (pendingAckMessageForCurrentTxn != null) {
+ for (Entry<PositionImpl, PositionImpl> entry : pendingAckMessageForCurrentTxn.entrySet()) {
+ if (pendingAckBatchMessageMap != null && pendingAckBatchMessageMap.containsKey(entry.getValue())) {
+ HashSet<TxnID> txnIDConcurrentOpenHashSet =
+ pendingAckBatchMessageMap.get(entry.getValue());
+ txnIDConcurrentOpenHashSet.remove(txnId);
+ if (txnIDConcurrentOpenHashSet.isEmpty()) {
+ pendingAckBatchMessageMap.remove(entry.getValue());
+ pendingAckMessages.remove(entry.getValue());
+ }
+ } else {
+ this.pendingAckMessages.remove(entry.getValue());
+ }
+ }
+ redeliverUnacknowledgedMessages(consumer, new ArrayList<>(pendingAckMessageForCurrentTxn.values()));
+ }
+ }
+ abortFuture.complete(null);
+ return abortFuture;
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer) {
+ ConcurrentLongLongPairHashMap positionMap = consumer.getPendingAcks();
+ // Only check if message is in pending_ack status when there's ongoing transaction.
+ if (null != positionMap && ((pendingAckMessages != null && pendingAckMessages.size() != 0)
+ || this.cumulativeAckPair != null)) {
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positionMap.asMap().forEach((key, value) -> {
+ PositionImpl position = new PositionImpl(key.first, key.second);
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition);
+ });
+
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer);
+ }
+ }
+
+ @Override
+ public void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
+ // If there's ongoing transaction.
+ if ((pendingAckMessages != null && pendingAckMessages.size() != 0) || this.cumulativeAckPair != null) {
+ // Check if message is in pending_ack status.
+ List<PositionImpl> pendingPositions = new ArrayList<>();
+ PositionImpl cumulativeAckPosition =
+ (null == this.cumulativeAckPair) ? null : this.cumulativeAckPair.getValue();
+
+ positions.forEach(position ->
+ redeliverUnacknowledgedMessagesCommon(position, pendingPositions, cumulativeAckPosition));
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, pendingPositions);
+ } else {
+ persistentSubscription.getDispatcher().redeliverUnacknowledgedMessages(consumer, positions);
+ }
+ }
+
+ private void redeliverUnacknowledgedMessagesCommon(PositionImpl position, List<PositionImpl> pendingPositions,
+ PositionImpl cumulativeAckPosition) {
+ if (position.hasAckSet()) {
+ if (this.pendingAckMessages != null) {
+ if (this.pendingAckMessages.containsKey(position)) {
+ if (!isAckSetRepeated(this.pendingAckMessages.get(position), position)) {
+ pendingPositions.add(position);
+ }
+ } else {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null) {
+ int compareNumber = position.compareTo((cumulativeAckPosition));
+ if (compareNumber > 0 || (compareNumber == 0 && !isAckSetRepeated(position, cumulativeAckPosition))) {
+ pendingPositions.add(position);
+ }
+ }
+ } else {
+ if (this.pendingAckMessages != null) {
+ if (!this.pendingAckMessages.containsKey(position)) {
+ pendingPositions.add(position);
+ }
+ } else if (cumulativeAckPosition != null ) {
+ if (position.compareTo(cumulativeAckPosition) > 0) {
+ pendingPositions.add(position);
+ }
+ }
+ }
+ }
+
+ public int compareToWithAckSetForCumulativeAck(PositionImpl currentPosition,PositionImpl otherPosition) {
+ int result = ComparisonChain.start().compare(currentPosition.getLedgerId(),
+ otherPosition.getLedgerId()).compare(currentPosition.getEntryId(), otherPosition.getEntryId())
+ .result();
+ if (result == 0) {
+ if (otherPosition.getAckSet() == null && currentPosition.getAckSet() == null) {
+ return result;
+ }
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(otherPosition.getAckSet());
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet());
+ if (otherAckSet.equals(thisAckSet)) {
+ return result;
+ }
+ otherAckSet.and(thisAckSet);
+ boolean flag = otherAckSet.equals(thisAckSet);
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return flag ? 1 : 0;
+ }
+ return result;
+ }
+
+ public boolean isAckSetRepeated(PositionImpl currentPosition, PositionImpl otherPosition) {
+ if (currentPosition.getAckSet() == null || otherPosition.getAckSet() == null) {
+ return false;
+ }
+
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet());
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(otherPosition.getAckSet());
+ if (otherAckSet.size() < thisAckSet.size()) {
+ otherAckSet.set(otherAckSet.size(), thisAckSet.size());
+ }
+ thisAckSet.flip(0, thisAckSet.size());
+ otherAckSet.flip(0, otherAckSet.size());
+ thisAckSet.and(otherAckSet);
+ boolean isAckSetRepeated = !thisAckSet.isEmpty();
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return isAckSetRepeated;
+ }
+
+ public void andAckSet(PositionImpl currentPosition, PositionImpl otherPosition) {
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(currentPosition.getAckSet());
Review comment:
Add check to ensure the ledger Id and the entry Id of the current position equals to the other position
And add a unit test for this method.
Is the method can be static?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513308918
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java
##########
@@ -265,8 +263,9 @@ public void testCanAcknowledgeAndAbortForTransaction() throws BrokerServiceExcep
persistentSubscription.acknowledgeMessage(txnID2, positions, AckType.Cumulative).get();
fail("Cumulative acknowledge for transaction2 should fail. ");
} catch (ExecutionException e) {
- assertEquals(e.getCause().getMessage(),"[persistent://prop/use/ns-abc/successTopic][subscriptionName] " +
- "Transaction:(1,2) try to cumulative ack message while transaction:(1,1) already cumulative acked messages.");
+ assertEquals(e.getCause().getMessage(),"[persistent://prop/use/ns-abc/successTopic]" +
+ "[subscriptionName] Transaction:(1,2) try to cumulative batch ack position: " +
+ "2:50 within range of current currentPosition: 1:100");
Review comment:
Assert the exception type?
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -272,6 +275,7 @@ private void txnAckTest(boolean batchEnable, int maxBatchSize,
// 2) ack committed by a new txn
commitTxn.commit().get();
+ Thread.sleep(2000L);
Review comment:
Why we need sleep here? Please check all.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
Review comment:
Add unit test for the PendingAckHandleImpl to ensure the cache that we maintain in the PendingAckHandleImpl can be removed correctly to avoid the memory leak.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ private Map<TxnID, HashMap<PositionImpl, PositionImpl>> pendingIndividualAckMessagesMap;
+
+ private Map<PositionImpl, PositionImpl> pendingAckMessages;
+
+ private Map<PositionImpl, HashSet<TxnID>> pendingAckBatchMessageMap;
+
+ private Pair<TxnID, PositionImpl> cumulativeAckPair;
Review comment:
Could you please add the detailed description for these structs?
1. What is it used for (when to insert and when to remove)
2. Give an example to show how the data maintained in the struct.
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
+ abortTxn.abort().get();
+ Thread.sleep(2000);
Review comment:
Any reason sleep here?
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
+ abortTxn.abort().get();
+ Thread.sleep(2000);
+ Transaction commitTxn = getTxn();
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), commitTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ commitTxn.commit().get();
+ Thread.sleep(2000);
+
+ try {
+ commitTxn.commit().get();
+ Assert.fail("recommit one transaction should be failed.");
+ } catch (Exception reCommitError) {
+ // recommit one transaction should be failed
+ log.info("expected exception for recommit one transaction.");
+ Assert.assertNotNull(reCommitError);
+ Assert.assertTrue(reCommitError.getCause() instanceof
+ TransactionCoordinatorClientException.InvalidTxnStatusException);
+ }
+
Review comment:
Same as above comment.
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
Review comment:
Check the following conditions
1. The ack should not conflict if the cumulative ack position is greater than the last acked
2. The ack should conflict if the cumulative ack position equals to the last acked
3. The ack should conflict if the cumulative ack position is lower than the last acked
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -261,6 +263,7 @@ private void txnAckTest(boolean batchEnable, int maxBatchSize,
// 1) txn abort
txn.abort().get();
+ Thread.sleep(2000L);
Review comment:
Why we need sleep here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-708861808
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511551054
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
Review comment:
for a moment, we only ensure one transaction can cumulative ack multiple times. if we only ensure cumulative ack position is greater than the old one, when the client disconnect, broker will dispatch the new message rather than the position witch was cumulative ack by the previous transaction. so, it has't consume the position with pervious transaction when the transaction abort.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-708810102
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715154
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
Review comment:
OK, I will do it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715029
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
Review comment:
it will completeException(NotAllowException). because of it will completeException(), the exception will message, so the String should be retain。
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511551054
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
Review comment:
for a moment, we only ensure one transaction can cumulative ack multiple times. if we only ensure cumulative ack position is greater than the old one, when the client disconnect, broker will dispatch the new message rather than the position witch was cumulative ack by the previous transaction. so, it has't consume the position with pervious transaction when the transaction abort.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513345226
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
Review comment:
1.
> The ack should not conflict if the cumulative ack position is greater than the last acked
```
if (i % 3 == 0) {
consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
}
```
it will check.
2.
> The ack should conflict if the cumulative ack position equals to the last acked
```
try {
consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
fail("not ack conflict ");
} catch (Exception e) {
Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
}
```
the message.getMessageId is equals the last acked.
3. because of the compare is <= 0, so we only check <=
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513351251
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,414 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import com.google.common.collect.ComparisonChain;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
Review comment:
OK
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r513335449
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -378,6 +382,91 @@ public void txnMessageAckTest() throws Exception {
log.info("receive transaction messages count: {}", receiveCnt);
}
+ @Test
+ public void txnAckTestBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(true, 200, SubscriptionType.Failover);
+ }
+
+ @Test
+ public void txnAckTestNoBatchAndCumulativeSub() throws Exception {
+ txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
+ }
+
+ public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ throws Exception {
+ String normalTopic = NAMESPACE1 + "/normal-topic";
+
+ @Cleanup
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(normalTopic)
+ .subscriptionName("test")
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subscriptionType)
+ .ackTimeout(1, TimeUnit.MINUTES)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(normalTopic)
+ .enableBatching(batchEnable)
+ .batchingMaxMessages(maxBatchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+
+ for (int retryCnt = 0; retryCnt < 2; retryCnt++) {
+ Transaction abortTxn = getTxn();
+ int messageCnt = 130;
+ // produce normal messages
+ for (int i = 0; i < messageCnt; i++){
+ producer.newMessage().value("hello".getBytes()).sendAsync();
+ }
+ Message<byte[]> message = null;
+ Thread.sleep(1000L);
+ for (int i = 0; i < messageCnt; i++) {
+ message = consumer.receive(1, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ if (i % 3 == 0) {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ }
+ log.info("receive msgId abort: {}, retryCount : {}, count : {}", message.getMessageId(), retryCnt, i);
+ }
+ try {
+ consumer.acknowledgeCumulativeAsync(message.getMessageId(), abortTxn).get();
+ fail("not ack conflict ");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
+ abortTxn.abort().get();
+ Thread.sleep(2000);
Review comment:
client commit txn and the TC will not call TB before, now it don't need to sleep.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r510573634
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
Review comment:
It's not a TransactionConflictException exception right? we must keep the exception handling clear. I think you can return a NotAllowedException.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
Review comment:
checkArgument will throw a RuntimeException, since the method returns `CompletableFuture<Void>`, it's better to complete the future with an exception.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
+ *
+ * @param txnId TransactionID of an ongoing transaction trying to sck message.
+ * @param positions {@link Position}(s) it try to ack.
+ * @param ackType {@link PulsarApi.CommandAck.AckType}.
+ * cumulative ack or try to single ack message already acked by any ongoing transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> acknowledgeMessage(TxnID txnId, List<Position> positions, PulsarApi.CommandAck.AckType ackType);
+
+ /**
+ * Commit a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param properties Additional user-defined properties that can be associated with a particular cursor position.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> commitTxn(TxnID txnId, Map<String,Long> properties);
+
+ /**
+ * Abort a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param consumer {@link Consumer} which aborting transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer);
+
+ void setPersistentSubscription(PersistentSubscription persistentSubscription);
Review comment:
Looks confused that introduces this method in the interface of PendingAckHandle. It's better to move it to the implementation internal.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
Review comment:
Is it related to the acknowledgeMessage method?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
Review comment:
Does this mean the one transaction only can allow one cumulative ack? I think we can allow a transaction to do cumulative ack multiple times, we only should ensure the new cumulative ack position is greater than the old one, right?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnId.toString(), position);
+ }
+ if (this.pendingCumulativeAckTxnId == null) {
+ // Only set pendingCumulativeAckTxnId if no transaction is doing cumulative ack.
+ PENDING_CUMULATIVE_ACK_TXNID_UPDATER.set(this, txnId);
+ POSITION_UPDATER.set(this, position);
Review comment:
There is no atomic guarantee for these two operations.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] TxnID:[{}] Cumulative ack on {}.", topicName, subName, txnId.toString(), position);
+ }
+ if (this.pendingCumulativeAckTxnId == null) {
+ // Only set pendingCumulativeAckTxnId if no transaction is doing cumulative ack.
+ PENDING_CUMULATIVE_ACK_TXNID_UPDATER.set(this, txnId);
+ POSITION_UPDATER.set(this, position);
+ } else if (this.pendingCumulativeAckTxnId.equals(txnId)
+ && position.compareToWithAckSet((PositionImpl) pendingCumulativeAckMessage) > 0) {
+ // Only set pendingCumulativeAckTxnId if no transaction is doing cumulative ack.
+ PENDING_CUMULATIVE_ACK_TXNID_UPDATER.set(this, txnId);
+ POSITION_UPDATER.set(this, position);
Review comment:
Same as above comment.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
Review comment:
Looks this method is complicated, you can split it into 2 parts, acknowledgeMessageIndividual and acknowledgeMessageCumulative.
```
if (AckType.Cumulative == ackType) {
acknowledgeMessageCumulative();
} else {
acknowledgeMessageIndividual();
}
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -331,7 +332,8 @@ private PersistentSubscription createPersistentSubscription(String subscriptionN
if (subscriptionName.equals(Compactor.COMPACTION_SUBSCRIPTION)) {
return new CompactorSubscription(this, compactedTopic, subscriptionName, cursor);
} else {
- return new PersistentSubscription(this, subscriptionName, cursor, replicated);
+ return new PersistentSubscription(this, subscriptionName, cursor,
+ replicated, new PendingAckHandleImpl(topic, subscriptionName));
Review comment:
The PendingAckHandle is an internal component of the subscription, so seems it's better to instantiate internal.
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
##########
@@ -121,6 +180,10 @@ public boolean equals(Object obj) {
return false;
}
+ public boolean isBatchPosition() {
+ return ackSet != null;
+ }
Review comment:
Do you mean `hasAckSet()`?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ if (position.compareTo((PositionImpl) persistentSubscription.getCursor().getMarkDeletedPosition()) <= 0) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " try to cumulative ack position: " + position + " within range of cursor's " +
+ "markDeletePosition: " + persistentSubscription.getCursor().getMarkDeletedPosition();
+ log.error(errorMsg);
Review comment:
It's easy to achieve by using log pattern log.error("[{}] [{}] Transaction:", topicName, subName).
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
+ return completableFuture;
+ }
+
+ checkArgument(positions.get(0) instanceof PositionImpl);
Review comment:
Please ensure the method doesn't throw any exception. Instead, complete the future with exception.
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
##########
@@ -81,6 +82,64 @@ public long getEntryId() {
return entryId;
}
+ public int compareToWithAckSet(PositionImpl other) {
+ checkNotNull(other);
+ int result = ComparisonChain.start().compare(this.ledgerId, other.ledgerId).compare(this.entryId, other.entryId)
+ .result();
+ if (result == 0) {
+ if (other.ackSet == null && ackSet == null) {
+ return result;
+ }
+ checkNotNull(other.ackSet);
+ checkNotNull(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ if (otherAckSet.equals(thisAckSet)) {
+ return result;
+ }
+ otherAckSet.and(thisAckSet);
+ boolean flag = otherAckSet.equals(thisAckSet);
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return flag ? 1 : 0;
+ }
+ return result;
+ }
+
+ public boolean isAckSetRepeated(PositionImpl other) {
+ if (ackSet == null || other.ackSet == null) {
+ return false;
+ }
+ checkNotNull(other);
+ checkNotNull(ackSet);
+ checkNotNull(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ if (otherAckSet.size() < thisAckSet.size()) {
+ otherAckSet.set(otherAckSet.size(), thisAckSet.size());
+ }
+ thisAckSet.flip(0, thisAckSet.size());
+ otherAckSet.flip(0, otherAckSet.size());
+ thisAckSet.and(otherAckSet);
+ boolean isAckSetRepeated = !thisAckSet.isEmpty();
+ thisAckSet.recycle();
+ otherAckSet.recycle();
+ return isAckSetRepeated;
+ }
+
+ public void andAckSet(PositionImpl other) {
+ checkNotNull(other);
+ checkNotNull(ackSet);
+ checkNotNull(other.ackSet);
+ BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(ackSet);
+ BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(other.ackSet);
+ if (otherAckSet.size() < thisAckSet.size()) {
+ otherAckSet.set(otherAckSet.size(), thisAckSet.size());
+ }
+ thisAckSet.and(otherAckSet);
+ this.ackSet = thisAckSet.toLongArray();
+ }
Review comment:
We do not need to introduce complex checking logic in PositionImpl and it is only used in the transaction. I think it's better to expose a setAckSet method if you want to update the ackSet.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
Review comment:
Do you mean allowed in one transaction? I think we should throw the ConflictException here since the transaction want to ack messages are already acked right?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
+ *
+ * @param txnId TransactionID of an ongoing transaction trying to sck message.
+ * @param positions {@link Position}(s) it try to ack.
+ * @param ackType {@link PulsarApi.CommandAck.AckType}.
+ * cumulative ack or try to single ack message already acked by any ongoing transaction.
Review comment:
Is it make sense here? Looks the comment does not relate to the `ackType`.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
+ *
+ * @param txnId TransactionID of an ongoing transaction trying to sck message.
+ * @param positions {@link Position}(s) it try to ack.
+ * @param ackType {@link PulsarApi.CommandAck.AckType}.
+ * cumulative ack or try to single ack message already acked by any ongoing transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> acknowledgeMessage(TxnID txnId, List<Position> positions, PulsarApi.CommandAck.AckType ackType);
+
+ /**
+ * Commit a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param properties Additional user-defined properties that can be associated with a particular cursor position.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> commitTxn(TxnID txnId, Map<String,Long> properties);
+
+ /**
+ * Abort a transaction.
+ *
+ * @param txnId {@link TxnID} to identify the transaction.
+ * @param consumer {@link Consumer} which aborting transaction.
+ * @return the future of this operation.
+ */
+ CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer);
+
+ void setPersistentSubscription(PersistentSubscription persistentSubscription);
+
+ void redeliverUnacknowledgedMessages(Consumer consumer);
+
+ void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions);
Review comment:
Could you please also add the method comment for these two methods?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
+ if (AckType.Cumulative == ackType) {
+
+ if (positions.size() != 1) {
+ String errorMsg = "[" + topicName + "][" + subName + "] Transaction:" + txnId +
+ " invalid cumulative ack received with multiple message ids.";
+ log.error(errorMsg);
+ completableFuture.completeExceptionally(new TransactionConflictException(errorMsg));
Review comment:
It's easy to achieve by using log pattern log.error("[{}] [{}] Transaction:", topicName, subName).
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
Review comment:
Same as above comment.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java
##########
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface PendingAckHandle {
+
+ /**
+ * Acknowledge message(s) for an ongoing transaction.
+ * <p>
+ * It can be of {@link PulsarApi.CommandAck.AckType#Individual} or {@link PulsarApi.CommandAck.AckType#Cumulative}. Single messages acked by ongoing
+ * transaction will be put in pending_ack state and only marked as deleted after transaction is committed.
+ * <p>
+ * Only one transaction is allowed to do cumulative ack on a subscription at a given time.
+ * If a transaction do multiple cumulative ack, only the one with largest position determined by
+ * {@link PositionImpl#compareTo(PositionImpl)} will be kept as it cover all position smaller than it.
+ * <p>
+ * If an ongoing transaction cumulative acked a message and then try to ack single message which is
+ * smaller than that one it cumulative acked, it'll succeed.
+ * <p>
+ * If transaction is aborted all messages acked by it will be put back to pending state.
+ *
+ * @param txnId TransactionID of an ongoing transaction trying to sck message.
+ * @param positions {@link Position}(s) it try to ack.
+ * @param ackType {@link PulsarApi.CommandAck.AckType}.
+ * cumulative ack or try to single ack message already acked by any ongoing transaction.
+ * @return the future of this operation.
Review comment:
It's better to clarify what exceptions will throw?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
Review comment:
Is it can be simplified by using a Pair<Txn, Position> ?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
Review comment:
And this will introduce a NPE here since the positions.get(0) may return null.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
+ this.topicName = topicName;
+ this.subName = subName;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> acknowledgeMessage(TxnID txnId,
+ List<Position> positions, AckType ackType) {
+ checkArgument(txnId != null, "TransactionID can not be null.");
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ checkArgument(positions.get(0) instanceof PositionImpl);
+ PositionImpl position = (PositionImpl) positions.get(0);
Review comment:
Same as the above comment.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
+
+ private PersistentSubscription persistentSubscription;
+
+ private final String topicName;
+
+ private final String subName;
+
+ public PendingAckHandleImpl(String topicName, String subName) {
Review comment:
You can instantiate with the PersistentSubscription and you can get topicName and subName from the PersistentSubscription
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511715534
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
##########
@@ -121,6 +180,10 @@ public boolean equals(Object obj) {
return false;
}
+ public boolean isBatchPosition() {
+ return ackSet != null;
+ }
Review comment:
yes, it has been modified
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r511716076
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -0,0 +1,363 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.pendingack.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
+import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class PendingAckHandleImpl implements PendingAckHandle {
+
+ // Map to keep track of message ack by each txn.
+ private ConcurrentOpenHashMap<TxnID, ConcurrentOpenHashMap<Position, Position>> pendingIndividualAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private ConcurrentOpenHashMap<Position, Position> pendingAckMessages;
+
+ private ConcurrentOpenHashMap<Position, ConcurrentOpenHashSet<TxnID>> pendingAckBatchMessageMap;
+
+ // Message cumulative acked by ongoing transaction, pending transaction commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, Position> POSITION_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, Position.class,
+ "pendingCumulativeAckMessage");
+
+ // ID of transaction currently using cumulative ack.
+ private volatile TxnID pendingCumulativeAckTxnId;
+
+ private static final AtomicReferenceFieldUpdater<PendingAckHandleImpl, TxnID> PENDING_CUMULATIVE_ACK_TXNID_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(PendingAckHandleImpl.class, TxnID.class,
+ "pendingCumulativeAckTxnId");
Review comment:
yes, it has been modified
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#discussion_r512008184
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -102,6 +102,7 @@
import org.apache.pulsar.broker.stats.NamespaceStats;
import org.apache.pulsar.broker.stats.ReplicationMetrics;
import org.apache.pulsar.broker.transaction.buffer.TransactionBuffer;
+import org.apache.pulsar.broker.transaction.pendingack.impl.PendingAckHandleImpl;
Review comment:
Seems we can delete the unused import?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui closed pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
codelipenghui closed pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-717631723
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-708811018
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 removed a comment on pull request #8256: Transaction pendingack server implement
Posted by GitBox <gi...@apache.org>.
congbobo184 removed a comment on pull request #8256:
URL: https://github.com/apache/pulsar/pull/8256#issuecomment-708810102
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org