You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "congbobo184 (via GitHub)" <gi...@apache.org> on 2023/04/06 15:07:34 UTC
[GitHub] [pulsar] congbobo184 opened a new pull request, #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch
congbobo184 opened a new pull request, #20032:
URL: https://github.com/apache/pulsar/pull/20032
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- Either this PR fixes an issue, -->
<!-- or this PR is one task of an issue -->
Master Issue:
Fixes client redeliver epoch bigger than broker consumer epoch.
Now redeliver method exists above race condition:
1. consumer reconnects to the broker with the epoch (1)
2. client consumer invokes redeliver command with epoch (2) and find the consumer doesn't connect, so ignore this redeliver command
3. the result is broker send message with epoch(1), client will filter these message
<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/wiki/proposals/PIP.md -->
### Motivation
fix this issue
### Modifications
1. client sends redeliver command don't check the consumer state only check the cnx to see whether been set to the client
2. broker consumer future complete, then process the redeliver command, if complete exception don't need to handle, because conusmer will reconnect with the epoch
### Verifying this change
add test for it
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### 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: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2459546477
> DEFAULT_CONSUMER_EPOCH as seem as CONSUMER_EPOCH.get(this), we need to solves the reconnect and redeliver race condition. Using DEFAULT_CONSUMER_EPOCH seems to make the problem more complicated
Ok, I can now see that the synchronized block is there to prevent the race in increasing the epoch. I hope that we'd have a proper test for the race by injecting a delay to have a real race.
There's also another issue with the permits. `increaseAvailablePermits` should only be called if writing of the redeliverUnacknowledgedMessages command succeeds. That should be done in the promise callback of writeAndFlush to ensure that permits aren't increased in the case where the connection is not available.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2488011880
> > > There's also another issue with the permits. `increaseAvailablePermits` should only be called if writing of the redeliverUnacknowledgedMessages command succeeds. That should be done in the promise callback of writeAndFlush to ensure that permits aren't increased in the case where the connection is not available.
> >
> >
> > yes, you are right. but this is a another problem, I think this pr don't need to handle this situation
>
> I agree that it's partially a different problem. However since this PR changes the behavior around it, I think that it would make sense to address the permit issue in this PR. Solving the issue will require a few lines of code.
again I see the code, "increaseAvailablePermits" is only relevant to "incomingMessages", no need to care about redeliverCommand success or failure.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1798876637
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2103,8 +2103,10 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // If a subscription command has been sent to the broker, it is necessary to allow the redelivery
+ // request to be sent to the broker without checking the connection state, as failing to do so would
+ // result in the client consumer epoch being bigger than the broker consumer epoch.
Review Comment:
Why does the connection state check need to be skipped? It's mentioned in the comment, but I don't understand the behavior.
--
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] Send the redeliver while the consumer reconnects so the broker learns the new epoch [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-6356863584
@congbobo184 this race is still on master, so I've taken the PR over: merged master and kept just the client send
plus the broker applying the redeliver once the consumer is created. I dropped the dispatcher change and the
`ServerCnx` refactor. The new test holds the topic so the subscribe stays pending and the race really happens
(lhotari's point); it fails without either half. While reconnecting, the client no longer adds permits for the
redeliver, since the subscribe success resets them. Thanks for finding this!
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2468836428
> > Ok, I can now see that the synchronized block is there to prevent the race in increasing the epoch. I hope that we'd have a proper test for the race by injecting a delay to have a real race in the test.
>
> Could you please find a similar test for me? I'm not very good at writing it.
I don't have a good example in mind. It might require a broader effort in introducing good ways to test race conditions in the client code. Optimally, the implementation would contain injection points for adding such delays.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1798874588
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -279,15 +279,21 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoc
}
private synchronized void internalRedeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoch) {
+ // broker side epoch is smaller than consumer epoch, so don't need to handle this redeliver request
+ if (consumerEpoch < consumer.getConsumerEpoch()) {
+ log.warn("[{}-{}] Ignoring redeliverUnacknowledgedMessages since broker epoch [{}] is smaller than "
+ + "consumer epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
+ return;
+ }
- if (consumerEpoch > consumer.getConsumerEpoch()) {
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Update epoch, old epoch [{}], new epoch [{}]",
- name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
- }
- consumer.setConsumerEpoch(consumerEpoch);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Update epoch, old epoch [{}] new epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
}
+ consumer.setConsumerEpoch(consumerEpoch);
Review Comment:
This doesn't make sense to me. Why is the broker side epoch updated 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.
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1830693798
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -279,15 +279,21 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoc
}
private synchronized void internalRedeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoch) {
+ // broker side epoch is smaller than consumer epoch, so don't need to handle this redeliver request
+ if (consumerEpoch < consumer.getConsumerEpoch()) {
+ log.warn("[{}-{}] Ignoring redeliverUnacknowledgedMessages since broker epoch [{}] is smaller than "
+ + "consumer epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
+ return;
+ }
- if (consumerEpoch > consumer.getConsumerEpoch()) {
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Update epoch, old epoch [{}], new epoch [{}]",
- name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
- }
- consumer.setConsumerEpoch(consumerEpoch);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Update epoch, old epoch [{}] new epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
}
+ consumer.setConsumerEpoch(consumerEpoch);
Review Comment:
I get it now. It seems that the code was just refactored in this PR and the logic wasn't really changed in this 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.
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1830668569
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -279,15 +279,21 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoc
}
private synchronized void internalRedeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoch) {
+ // broker side epoch is smaller than consumer epoch, so don't need to handle this redeliver request
+ if (consumerEpoch < consumer.getConsumerEpoch()) {
+ log.warn("[{}-{}] Ignoring redeliverUnacknowledgedMessages since broker epoch [{}] is smaller than "
+ + "consumer epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
+ return;
+ }
- if (consumerEpoch > consumer.getConsumerEpoch()) {
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Update epoch, old epoch [{}], new epoch [{}]",
- name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
- }
- consumer.setConsumerEpoch(consumerEpoch);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Update epoch, old epoch [{}] new epoch [{}]",
+ name, consumer, consumer.getConsumerEpoch(), consumerEpoch);
}
+ consumer.setConsumerEpoch(consumerEpoch);
Review Comment:
this means that client consumer invoke redeliverUnacknowledgedMessages, client consumer hope that broker can next send messages with consumerEpoch.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2103,8 +2103,10 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // If a subscription command has been sent to the broker, it is necessary to allow the redelivery
+ // request to be sent to the broker without checking the connection state, as failing to do so would
+ // result in the client consumer epoch being bigger than the broker consumer epoch.
Review Comment:
- client consumer send reconnect command to broker epoch = 1
- client consumer invoke redeliverUnacknowledgedMessages epoch = 2
- if check connect, client consumer will not send redeliverUnacknowledgedMessages command with epoch = 2 to broker
- client consumer epoch = 2 will bigger than broker consumer epoch = 2, client will filter all messages with epoch = 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
[GitHub] [pulsar] github-actions[bot] commented on pull request #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-1537267601
The pr had no activity for 30 days, mark with Stale label.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2459498198
> @congbobo184 To me it seems that the problem could be prevented by changing the this line
>
> https://github.com/apache/pulsar/blob/8eeb0e2e89f8938348d29044d9e1d843a6251067/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L898
>
>
> I'm assuming that instead of passing `CONSUMER_EPOCH.get(this)`, using `DEFAULT_CONSUMER_EPOCH` would solve the issue. It might also require changes so that DEFAULT_CONSUMER_EPOCH would always redeliver messages on the broker side. The client side seems to already always accept DEFAULT_CONSUMER_EPOCH.
> Do you agree that this would solve the problem?
DEFAULT_CONSUMER_EPOCH as seem as CONSUMER_EPOCH.get(this), we need to solves the reconnect and redeliver race condition, Using DEFAULT_CONSUMER_EPOCH seems to make the problem more complicated
--
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] Send the redeliver while the consumer reconnects so the broker learns the new epoch [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r4360746429
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2447,11 +2447,14 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // Send the redeliver also while the subscribe is in flight on this cnx: it carried the previous
+ // epoch, and if the broker never learns the new one, every message it dispatches is discarded here.
+ // The broker applies the redeliver once the consumer is created.
+ if (cnx != null) {
cnx.ctx().writeAndFlush(Commands.newRedeliverUnacknowledgedMessages(
Review Comment:
Good catch, thanks. Confirmed: when the reconnection reuses a pooled connection, `connectionOpened` runs on the
lookup thread, so the SUBSCRIBE write is queued to the event loop. A redeliver called from a callback on that
event loop, such as a producer send receipt, writes immediately and gets ahead of it. I went with your suggestion
and made the fix client-only. The consumer keeps the epoch it subscribed with. Once the subscribe succeeds, if
the epoch has moved, it sends the current one directly, before `consumerIsReconnectedToBroker` and the initial
flow. That covers every ordering, so I reverted the early send while `Connecting`, which also avoids sending
the same epoch twice.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java:
##########
@@ -321,6 +326,51 @@ public void testRedeliveryAddEpoch(boolean enableBatch) throws Exception{
assertEquals(message.getValue(), test3);
}
+ /**
+ * A redeliver called while the consumer reconnects bumps the client epoch after the subscribe was sent with
+ * the previous one. The broker must still learn the new epoch, or the client discards every message.
+ */
+ @Test(timeOut = 60_000)
+ public void testRedeliveryWhileReconnecting() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testRedeliveryWhileReconnecting";
+ final String subName = "my-sub";
+ // A client of its own, so that the consumer's connection only carries the consumer's subscribe
+ @Cleanup
+ PulsarClient consumerClient = newPulsarClient(lookupUrl.toString(), 0);
+ @Cleanup
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) consumerClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe();
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create()) {
+ producer.send("Pulsar");
+ }
+
+ // Hold the topic, so that the subscribe sent by the reconnection stays pending on the broker
+ Map<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService().getTopics();
+ CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
+ CompletableFuture<Optional<Topic>> heldTopicFuture = new CompletableFuture<>();
+ topics.put(topic, heldTopicFuture);
+ try {
+ consumer.getClientCnx().channel().close();
+ Awaitility.await().until(() -> {
+ ClientCnx cnx = consumer.getClientCnx();
+ return consumer.getState() == HandlerState.State.Connecting && cnx != null
+ && !cnx.pendingRequests.isEmpty();
+ });
+
+ consumer.redeliverUnacknowledgedMessages();
+ } finally {
+ topics.put(topic, topicFuture);
+ heldTopicFuture.complete(topicFuture.join());
Review Comment:
Agreed: with the broker change, the test depended on timing. The broker change is gone now; the client-side
check doesn't need it, and it works with any broker version. The client no longer sends anything while
`Connecting`, so the broker can only learn the new epoch from the send after the subscribe succeeds. The test
fails without it (checked by reverting `ConsumerImpl`): nothing else sends the new epoch.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1830658479
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2103,8 +2103,10 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // If a subscription command has been sent to the broker, it is necessary to allow the redelivery
+ // request to be sent to the broker without checking the connection state, as failing to do so would
+ // result in the client consumer epoch being bigger than the broker consumer epoch.
Review Comment:
- client consumer send reconnect command to broker epoch = 1
- client consumer invoke redeliverUnacknowledgedMessages epoch = 2
- if check connect(when state is not ready), client consumer will not send redeliverUnacknowledgedMessages command with epoch = 2 to broker
- client consumer epoch = 2 will bigger than broker consumer epoch = 1, client will filter all messages with epoch = 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
Re: [PR] [fix][client] Fix client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2468830783
> > There's also another issue with the permits. `increaseAvailablePermits` should only be called if writing of the redeliverUnacknowledgedMessages command succeeds. That should be done in the promise callback of writeAndFlush to ensure that permits aren't increased in the case where the connection is not available.
>
> yes, you are right. but this is a another problem, I think this pr don't need to handle this situation
I agree that it's partially a different problem. However since this PR changes the behavior around it, I think that it would make sense to address the permit issue in this PR. Solving the issue will require a few lines of code.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari closed pull request #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch
URL: https://github.com/apache/pulsar/pull/20032
--
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
[GitHub] [pulsar] lhotari commented on a diff in pull request #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1189700517
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1812,20 +1812,31 @@ protected void handleRedeliverUnacknowledged(CommandRedeliverUnacknowledgedMessa
remoteAddress, redeliver.getConsumerId(),
redeliver.hasConsumerEpoch() ? redeliver.getConsumerEpoch() : null);
}
-
- CompletableFuture<Consumer> consumerFuture = consumers.get(redeliver.getConsumerId());
-
- if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
- Consumer consumer = consumerFuture.getNow(null);
- if (redeliver.getMessageIdsCount() > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
- consumer.redeliverUnacknowledgedMessages(redeliver.getMessageIdsList());
- } else {
- if (redeliver.hasConsumerEpoch()) {
- consumer.redeliverUnacknowledgedMessages(redeliver.getConsumerEpoch());
+ boolean hasConsumerEpoch = redeliver.hasConsumerEpoch();
+ List<MessageIdData> messageIdsList = redeliver.getMessageIdsList();
+ int messageIdsCount = redeliver.getMessageIdsCount();
+ long consumerId = redeliver.getConsumerId();
+ long consumerEpoch = redeliver.getConsumerEpoch();
+ CompletableFuture<Consumer> consumerFuture = consumers.get(consumerId);
+ if (consumerFuture != null) {
+ consumerFuture.thenAccept((consumer) -> {
+ if (messageIdsCount > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
+ consumer.redeliverUnacknowledgedMessages(messageIdsList);
} else {
- consumer.redeliverUnacknowledgedMessages(DEFAULT_CONSUMER_EPOCH);
+ if (hasConsumerEpoch) {
+ consumer.redeliverUnacknowledgedMessages(consumerEpoch);
+ } else {
+ consumer.redeliverUnacknowledgedMessages(DEFAULT_CONSUMER_EPOCH);
+ }
}
- }
+ }).exceptionally(e -> {
+ // if consumerFuture completed exceptionally, don't need to process this redeliver command
+ // because, consumer will reconnect
+ log.warn("[{}] ignore this redeliverUnacknowledged request from consumer {}, consumerEpoch {}",
+ remoteAddress, redeliver.getConsumerId(),
+ redeliver.hasConsumerEpoch() ? redeliver.getConsumerEpoch() : null, e);
+ return null;
+ });
Review Comment:
Isn't this refactoring completely irrelevant in this PR? If the intention is to add a log statement, just add that without refactoring the existing logic.
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2410180355
closing and reopening to trigger CI
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2459237992
@congbobo184 To me it seems that the problem could be prevented by changing the this line https://github.com/apache/pulsar/blob/8eeb0e2e89f8938348d29044d9e1d843a6251067/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L898
I'm assuming that instead of passing `CONSUMER_EPOCH.get(this)`, using `DEFAULT_CONSUMER_EPOCH` would solve the issue. Do you agree?
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#issuecomment-2467181435
> > DEFAULT_CONSUMER_EPOCH as seem as CONSUMER_EPOCH.get(this), we need to solves the reconnect and redeliver race condition. Using DEFAULT_CONSUMER_EPOCH seems to make the problem more complicated
>
> Ok, I can now see that the synchronized block is there to prevent the race in increasing the epoch. I hope that we'd have a proper test for the race by injecting a delay to have a real race in the test.
Could you please find a similar test? I'm not very good at writing it.
>
> There's also another issue with the permits. `increaseAvailablePermits` should only be called if writing of the redeliverUnacknowledgedMessages command succeeds. That should be done in the promise callback of writeAndFlush to ensure that permits aren't increased in the case where the connection is not available.
yes, you are right. but this is a another problem, I think this pr don't need to handle this situation
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari closed pull request #20032: [fix][client] Fix client redeliver epoch bigger than broker consumer epoch
URL: https://github.com/apache/pulsar/pull/20032
--
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 client redeliver epoch bigger than broker consumer epoch [pulsar]
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r1830658479
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2103,8 +2103,10 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // If a subscription command has been sent to the broker, it is necessary to allow the redelivery
+ // request to be sent to the broker without checking the connection state, as failing to do so would
+ // result in the client consumer epoch being bigger than the broker consumer epoch.
Review Comment:
- client consumer send reconnect command to broker epoch = 1
- client consumer invoke redeliverUnacknowledgedMessages epoch = 2
- if check connect, client consumer will not send redeliverUnacknowledgedMessages command with epoch = 2 to broker
- client consumer epoch = 2 will bigger than broker consumer epoch = 1, client will filter all messages with epoch = 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
Re: [PR] [fix][client] Fix consumer discarding all messages after redelivering during a reconnection [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #20032:
URL: https://github.com/apache/pulsar/pull/20032
--
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] Send the redeliver while the consumer reconnects so the broker learns the new epoch [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #20032:
URL: https://github.com/apache/pulsar/pull/20032#discussion_r4359529028
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -2447,11 +2447,14 @@ public void redeliverUnacknowledgedMessages() {
incomingQueueLock.unlock();
}
- // is channel is connected, we should send redeliver command to broker
- if (cnx != null && isConnected(cnx)) {
+ // Send the redeliver also while the subscribe is in flight on this cnx: it carried the previous
+ // epoch, and if the broker never learns the new one, every message it dispatches is discarded here.
+ // The broker applies the redeliver once the consumer is created.
+ if (cnx != null) {
cnx.ctx().writeAndFlush(Commands.newRedeliverUnacknowledgedMessages(
Review Comment:
Thanks for addressing this race. There is still a possible ordering issue when a reconnect reuses an existing connection.
The thread handling the lookup can queue `SUBSCRIBE(epoch=0)` for the connection’s IO thread. Before that queued write runs, a producer callback on the IO thread can request redelivery and write `REDELIVER(epoch=1)` immediately.
The broker then receives REDELIVER first and ignores it because the consumer does not exist yet. It later receives SUBSCRIBE with epoch 0, while the client is already at epoch 1. Messages from that subscription would still be filtered by the client.
Checking for an epoch change after SUBSCRIBE succeeds, and sending the latest epoch before the reconnect hook or any FLOW, would cover this ordering too. Sending the command directly would avoid incrementing the epoch again.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java:
##########
@@ -321,6 +326,51 @@ public void testRedeliveryAddEpoch(boolean enableBatch) throws Exception{
assertEquals(message.getValue(), test3);
}
+ /**
+ * A redeliver called while the consumer reconnects bumps the client epoch after the subscribe was sent with
+ * the previous one. The broker must still learn the new epoch, or the client discards every message.
+ */
+ @Test(timeOut = 60_000)
+ public void testRedeliveryWhileReconnecting() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testRedeliveryWhileReconnecting";
+ final String subName = "my-sub";
+ // A client of its own, so that the consumer's connection only carries the consumer's subscribe
+ @Cleanup
+ PulsarClient consumerClient = newPulsarClient(lookupUrl.toString(), 0);
+ @Cleanup
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) consumerClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe();
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create()) {
+ producer.send("Pulsar");
+ }
+
+ // Hold the topic, so that the subscribe sent by the reconnection stays pending on the broker
+ Map<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService().getTopics();
+ CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
+ CompletableFuture<Optional<Topic>> heldTopicFuture = new CompletableFuture<>();
+ topics.put(topic, heldTopicFuture);
+ try {
+ consumer.getClientCnx().channel().close();
+ Awaitility.await().until(() -> {
+ ClientCnx cnx = consumer.getClientCnx();
+ return consumer.getState() == HandlerState.State.Connecting && cnx != null
+ && !cnx.pendingRequests.isEmpty();
+ });
+
+ consumer.redeliverUnacknowledgedMessages();
+ } finally {
+ topics.put(topic, topicFuture);
+ heldTopicFuture.complete(topicFuture.join());
Review Comment:
This test makes the client request redelivery while SUBSCRIBE is pending. However, returning from `redeliverUnacknowledgedMessages()` does not mean the broker has received the command yet.
The test immediately releases the topic, so the broker may finish creating the consumer before REDELIVER arrives. In that order, the old broker code can already handle the command, and the test can pass without exercising the new branch.
If the broker-side change remains, waiting until the broker has handled REDELIVER while consumer creation is still pending, and only then releasing the topic, would ensure that the test covers this branch.
--
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