You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2024/01/18 17:53:00 UTC
[PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
poorbarcode opened a new pull request, #21928:
URL: https://github.com/apache/pulsar/pull/21928
### Motivation
The `ack` command will fail when the consumer is reconnecting when setting `acknowledgmentGroupTime ->0`. see https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java#L361-L364
```java
private CompletableFuture<Void> doImmediateAck(MessageIdAdv msgId, AckType ackType, Map<String, Long> properties,
BitSetRecyclable bitSet) {
ClientCnx cnx = consumer.getClientCnx();
if (cnx == null) {
return FutureUtil.failedFuture(new PulsarClientException
.ConnectException("Consumer connect fail! consumer state:" + consumer.getState()));
}
return newImmediateAckAndFlush(consumer.consumerId, msgId, bitSet, ackType, properties, cnx);
}
```
### Modifications
Call `ack` after the consumer is connected.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Queue acks without grouping while the consumer is reconnecting [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #21928:
URL: https://github.com/apache/pulsar/pull/21928
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Queue acks without grouping while the consumer is reconnecting [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#issuecomment-6353595099
@poorbarcode this is still reproducible on master: with `acknowledgmentGroupTime(0)`, `doImmediateAck` fails when the consumer is reconnecting, and the V5 client drops that failure silently. Since you're busy elsewhere, I've picked it up here. I merged master and re-implemented the fix along the lines of @BewareMyPower's suggestion. Acks without properties go through the regular grouping path while the consumer isn't subscribed, and they're flushed once the subscribe succeeds. This also covers acks sent on the new connection before the subscribe response arrives, which the broker drops. I've updated the title and description accordingly. PTAL @BewareMyPower @codelipenghui
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1458193299
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java:
##########
@@ -4692,4 +4696,57 @@ public void flush(ChannelHandlerContext ctx) throws Exception {
consumer.close();
admin.topics().delete(topic, false);
}
+
+ @DataProvider(name = "ackArgs")
+ public Object[] ackArgs() {
+ int batchSize = 10;
+ BitSet bitSet = new BitSet(batchSize);
+ bitSet.set(0);
+ return new Object[][]{
+ // no batch.
+ {CommandAck.AckType.Cumulative, new MessageIdImpl(1,1,1)},
+ {CommandAck.AckType.Individual, new MessageIdImpl(1,1,1)},
+ // batch without ackSet.
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0)},
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0)},
+ // batch with ackSe.
Review Comment:
```suggestion
// batch with ackSet.
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java:
##########
@@ -4692,4 +4696,57 @@ public void flush(ChannelHandlerContext ctx) throws Exception {
consumer.close();
admin.topics().delete(topic, false);
}
+
+ @DataProvider(name = "ackArgs")
+ public Object[] ackArgs() {
+ int batchSize = 10;
+ BitSet bitSet = new BitSet(batchSize);
+ bitSet.set(0);
+ return new Object[][]{
+ // no batch.
+ {CommandAck.AckType.Cumulative, new MessageIdImpl(1,1,1)},
+ {CommandAck.AckType.Individual, new MessageIdImpl(1,1,1)},
+ // batch without ackSet.
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0)},
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0)},
+ // batch with ackSe.
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0, batchSize, bitSet)},
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0, batchSize, bitSet)}
+ };
+ }
+
+ @Test(dataProvider = "ackArgs")
+ public void testImmediateAckWhenReconnecting(CommandAck.AckType ackType, MessageId messageId) throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp_");
+ final String subscriptionName = "s1";
+ PulsarClient delayConnectClient = createDelayReconnectClient();
Review Comment:
The client should be closed?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
Will it break the behavior of
```
// We cannot group acks if the delay is 0 or when there are properties attached to it. Fortunately that's an
// uncommon condition since it's only used for the compaction subscription.
```
If the consumer is reconnecting but the ack has properties. We will also group the acks which is not expected?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -871,6 +871,7 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
}
+ acknowledgmentsGroupingTracker.afterConsumerReconnected();
Review Comment:
Why not call `flush()` method directly?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1458251277
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
@codelipenghui Yes. But this behavior is only added for the `TwoPhaseCompactor` to avoid latency when seeking, the client side has no way to attach properties. see https://github.com/apache/pulsar/blob/55520bd810d29991d4af2124151c8d75c25b0800/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java#L229-L230
Currently, if `acknowledge` is called during reconnection, `phaseTwoSeekThenLoop` will fail.
From my perspective, we should also queue the ACK requests and flush them after connected. @poorbarcode
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461095649
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
Review Comment:
Sure, I will change the implementation of this PR as https://github.com/apache/pulsar/pull/21928#discussion_r1461093751
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461093751
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
Agress with @BewareMyPower
I am trying to split this PR into the following.
- part-1: add a new component `AcknowledgmentCache` to support caching the acknowledgments which include the arg `properties.`
- part-2: split `PersistentAcknowledgmentsGroupingTracker` into two implementations:
- cache and batch the acks
- immediately ack
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1458251277
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
@codelipenghui Yes. But this behavior is only added for the `TwoPhaseCompactor`, the client side has no way to attach properties. see https://github.com/apache/pulsar/blob/55520bd810d29991d4af2124151c8d75c25b0800/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java#L229-L230
Currently, if `acknowledge` is called during reconnection, `phaseTwoSeekThenLoop` will fail.
From my perspective, we should also queue the ACK requests and flush them after connected. @poorbarcode
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461256290
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
> split PersistentAcknowledgmentsGroupingTracker into two implementations
It's reasonable. BTW, the design of the C++ client is better that:
- [AckGroupingEnabled](https://github.com/apache/pulsar-client-cpp/blob/main/lib/AckGroupingTrackerEnabled.cc): cache the acknowledgments and flush them due to the timeout or count
- [AckGrouingDisabled](https://github.com/apache/pulsar-client-cpp/blob/main/lib/AckGroupingTrackerDisabled.cc): acknowledge immediately
https://github.com/apache/pulsar-client-cpp/blob/72b7311aeef32e28a28e926da686aaf948e8f948/lib/ConsumerImpl.cc#L201C1-L215C6
Though the naming is not good (but consistent with other impl classes in the library)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#issuecomment-2410978459
Please rebase
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461256290
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
> split PersistentAcknowledgmentsGroupingTracker into two implementations
It's reasonable. BTW, the C++ client is already implemented in such way that:
- [AckGroupingEnabled](https://github.com/apache/pulsar-client-cpp/blob/main/lib/AckGroupingTrackerEnabled.cc): cache the acknowledgments and flush them due to the timeout or count
- [AckGrouingDisabled](https://github.com/apache/pulsar-client-cpp/blob/main/lib/AckGroupingTrackerDisabled.cc): acknowledge immediately
https://github.com/apache/pulsar-client-cpp/blob/72b7311aeef32e28a28e926da686aaf948e8f948/lib/ConsumerImpl.cc#L201C1-L215C6
Though the naming is not good (but consistent with other impl classes in the library)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461093751
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
Agress with @BewareMyPower
I am trying to split this PR into the following.
- part-1: add a new component `AcknowledgmentCache` to support caching the acknowledgments which include the arg `properties.`
- part-2: split `PersistentAcknowledgmentsGroupingTracker` into two implementations:
- cache and batch the acks
- immediately ack
- part-3: fix the issue
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461095814
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
Marked current PR as `Draft`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#issuecomment-2494263354
@poorbarcode Please rebase
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1458247191
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
Review Comment:
Method invocations like `f(args..., true)` and `f(args..., false)` are hard to read unless you jump to the implementation of `f` and see what the boolean parameter means.
You should add a new method like `queueIndividualAck` and call it directly. For example,
```java
private CompletableFuture<Void> queueIndividualAck(MessageIdAdv messageId) {
Optional<Lock> readLock = acquireReadLock();
try {
doIndividualAckAsync(messageId);
return readLock.map(__ -> currentIndividualAckFuture).orElse(CompletableFuture.completedFuture(null));
} finally {
readLock.ifPresent(Lock::unlock);
if (pendingIndividualAcks.size() >= maxAckGroupSize) {
flush();
}
}
}
private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
// We cannot group acks if the delay is 0 or when there are properties attached to it. Fortunately that's an
// uncommon condition since it's only used for the compaction subscription.
return doImmediateAck(messageId, AckType.Individual, properties, null);
} else {
return queueIndividualAck(messageId);
}
}
```
Then you don't have to add `false` argument to all existing `doIndividualAck` calls. And in `doImmediateAck`, you can call `queueIndividualAck` directly.
```java
private CompletableFuture<Void> doImmediateAck(MessageIdAdv msgId, AckType ackType, Map<String, Long> properties,
BitSetRecyclable bitSet) {
ClientCnx cnx = consumer.getClientCnx();
if (cnx == null && consumer.getState() == HandlerState.State.Connecting) {
if (ackType == AckType.Cumulative) {
return queueCumulativeAck(msgId);
} else {
return queueIndividualAck(msgId);
}
}
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461095231
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java:
##########
@@ -4692,4 +4696,57 @@ public void flush(ChannelHandlerContext ctx) throws Exception {
consumer.close();
admin.topics().delete(topic, false);
}
+
+ @DataProvider(name = "ackArgs")
+ public Object[] ackArgs() {
+ int batchSize = 10;
+ BitSet bitSet = new BitSet(batchSize);
+ bitSet.set(0);
+ return new Object[][]{
+ // no batch.
+ {CommandAck.AckType.Cumulative, new MessageIdImpl(1,1,1)},
+ {CommandAck.AckType.Individual, new MessageIdImpl(1,1,1)},
+ // batch without ackSet.
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0)},
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0)},
+ // batch with ackSe.
Review Comment:
Fixed
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java:
##########
@@ -4692,4 +4696,57 @@ public void flush(ChannelHandlerContext ctx) throws Exception {
consumer.close();
admin.topics().delete(topic, false);
}
+
+ @DataProvider(name = "ackArgs")
+ public Object[] ackArgs() {
+ int batchSize = 10;
+ BitSet bitSet = new BitSet(batchSize);
+ bitSet.set(0);
+ return new Object[][]{
+ // no batch.
+ {CommandAck.AckType.Cumulative, new MessageIdImpl(1,1,1)},
+ {CommandAck.AckType.Individual, new MessageIdImpl(1,1,1)},
+ // batch without ackSet.
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0)},
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0)},
+ // batch with ackSe.
+ {CommandAck.AckType.Cumulative, new BatchMessageIdImpl(1,1,1,0, batchSize, bitSet)},
+ {CommandAck.AckType.Individual, new BatchMessageIdImpl(1,1,1,0, batchSize, bitSet)}
+ };
+ }
+
+ @Test(dataProvider = "ackArgs")
+ public void testImmediateAckWhenReconnecting(CommandAck.AckType ackType, MessageId messageId) throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp_");
+ final String subscriptionName = "s1";
+ PulsarClient delayConnectClient = createDelayReconnectClient();
Review Comment:
Fixed
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -871,6 +871,7 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
}
+ acknowledgmentsGroupingTracker.afterConsumerReconnected();
Review Comment:
Fixed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1461093751
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -231,29 +231,31 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
case Individual:
return addIndividualAcknowledgment(msgId,
batchMessageId,
- __ -> doIndividualAck(__, properties),
- __ -> doIndividualBatchAck(__, properties));
+ __ -> doIndividualAck(__, properties, false),
+ __ -> doIndividualBatchAck(__, properties, false));
case Cumulative:
if (batchMessageId != null) {
consumer.onAcknowledgeCumulative(batchMessageId, null);
} else {
consumer.onAcknowledgeCumulative(msgId, null);
}
if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
- return doCumulativeAck(msgId, properties, null);
+ return doCumulativeAck(msgId, properties, null, false);
} else if (batchIndexAckEnabled) {
- return doCumulativeBatchIndexAck(batchMessageId, properties);
+ return doCumulativeBatchIndexAck(batchMessageId, properties, false);
} else {
- doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null);
+ doCumulativeAck(MessageIdAdvUtils.prevMessageId(batchMessageId), properties, null, false);
return CompletableFuture.completedFuture(null);
}
default:
throw new IllegalStateException("Unknown AckType: " + ackType);
}
}
- private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
+ private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties,
+ boolean queueDueToConnecting) {
+ if (!queueDueToConnecting
+ && (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty()))) {
Review Comment:
Agress with @BewareMyPower
I am trying to split this PR into the following.
- add a new component `AcknowledgmentCache` to support caching the acknowledgments which include the arg `properties.`
- split `PersistentAcknowledgmentsGroupingTracker` into two implementations:
- cache and batch the acks
- immediately ack
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] fix ack failed when consumer is reconnecting [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21928:
URL: https://github.com/apache/pulsar/pull/21928#discussion_r1458239527
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -871,6 +871,7 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
}
+ acknowledgmentsGroupingTracker.afterConsumerReconnected();
Review Comment:
+1
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org