[fix][client] Fix client redeliver epoch bigger than broker consumer epoch

[congbobo184]

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

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:

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

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:
congbobo184#16

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[lhotari]

LGTM

[lhotari]

closing and reopening to trigger CI

[lhotari]

Please check the question about updating the consumer epoch

[lhotari]

Thanks for the explanations. Now I see the problem. It seems that there's a broader possibility for race conditions in all redeliverUnacknowledgedMessages methods. I think that this needs a different approach for fixing this and testing this issue.
There are multiple chances for the race to happen. I'll think of possible solutions.
The problem in the test case added in this PR is that it's a synthetic test where the internal state is modified. Instead of doing that, it would be better to have a way to introduce the race condition by having a way to inject a delay in the client side connection logic so that the race condition actually happens. I'll follow up with more details, possibly after experimenting on this.

[lhotari]

@congbobo184 To me it seems that the problem could be prevented by changing the this line

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 898 in 8eeb0e2

conf.getSubscriptionProperties(), CONSUMER_EPOCH.get(this));

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?

[congbobo184]

@congbobo184 To me it seems that the problem could be prevented by changing the this line

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 898 in 8eeb0e2

conf.getSubscriptionProperties(), CONSUMER_EPOCH.get(this));

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

[lhotari]

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.

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.

[congbobo184]

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 for me? 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

[lhotari]

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.

[lhotari]

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.

[congbobo184]

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.