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:
[email protected]
