Fixed reader cursor seek with disconnection

[merlimat]

Motivation

Currently the seek() operation will first disconnect all the connected consumers and then reset the cursor. This was due initially to simplify the logic and avoid concurrent operations.

Adding it now, so that consumers won't get disconnected/reconnected while the seek is performed.

[codelipenghui]

@merlimat I have check the logic but not found how to handle that reset cursor to a historical position and then consumer call acknowledgeCumulative() with messageId which is dispatched to this consumer before. If disconnect all consumers while reset cursor, client will create new consumers, that's not going to happen. Maybe i missed something.

[codelipenghui]

retest this please

[codelipenghui]

run java8 tests

[sijie]

@merlimat If we don't disconnect the consumers, how do consumers reset their state? e.g. pending queue, acknowledgement, and etc.

[merlimat]

If we don't disconnect the consumers, how do consumers reset their state? e.g. pending queue, acknowledgement, and etc.

Consumers they won't reset, they just will see the messages restart from an earlier (or later) point in time. I don't believe that should affect the consumer logic since there will be no visible disruption from consumer perspective.

[merlimat]

If disconnect all consumers while reset cursor, client will create new consumers, that's not going to happen

It can still happen, and actually it's more probable to happen the longer the ack is delayed from the receival of the message.

Even if the consumer session is recreated, a consumer can receive a message, then seek it's triggered and then it will call ack-cumulative returning the cursor to the previous position.

An application that seeks while acking (either individual or cumulative) is bound to unpredictable behavior in any case, irrespective of the disconnection.

[wolfstudy]

@merlimat I changed the milestone to 2.4.2.

[jiazhai]

seems the test failures related with this change:

org.apache.pulsar.compaction.CompactionTest.testCompactorReadsCompacted
org.apache.pulsar.compaction.CompactionTest.testSeekEarliestAfterCompaction
org.apache.pulsar.compaction.CompactorTest.testCompactAddCompact

[jiazhai]

If we don't disconnect the consumers, how do consumers reset their state? e.g. pending queue, acknowledgement, and etc.

Consumers they won't reset, they just will see the messages restart from an earlier (or later) point in time. I don't believe that should affect the consumer logic since there will be no visible disruption from consumer perspective.

Maybe Sijie was concern about the messages that already cached in consumer client. We may need some handling to clean the cached messages?

[sijie]

@merlimat I wasn't sure how is the ordering guaranteed if we don't clean up the cached message. Can you clarify it?

[sijie]

Moved this to 2.5.0 since I think it requires some more discussions about the side effect of this change.

[sijie]

actually moved to 2.4.3

[sijie]

@bsideup thank you for your updates. Do these N sessions belong to one subscription?

[bsideup]

@sijie yes, they listen to the same topic / partition with failover

[sijie]

@bsideup sorry. if they are in the same subscription, it is expected that all consumers in the same subscription to reset to the new cursor position if the cursor is reset from a correctness perspective. I see what can be improved here is when the broker disconnects the consumers, it can disconnect the active consumers only. For exclusive/failover subscription, it will only reset the consumer that is currently active. This seems to be a correct fix. @codelipenghui @merlimat what do you think?

[codelipenghui]

@merlimat I move it to 2.7.0 and feel free to move it back if you want to onboard it in 2.6.0.

[bsideup]

@codelipenghui it is sad to see it being moved over and over 😞

It is a blocker for failover subscriptions with external offset management, and the implemented behaviour is rather questionable.

What's the reason to move it (again)?

[codelipenghui]

@bsideup Thanks for your feedback, I didn't notice that it is a blocker before. I will move it back to 2.6.0 and could you please take a look at sijie's last comment? Is the approach works for you?

[codelipenghui]

@bsideup I have pushed a PR #7141 which based on sijie's comment #5278 (comment). Could you please help review the changes?

[bsideup]

@codelipenghui looks promising! Looking forward to trying it 👍

[codelipenghui]

@bsideup PR #7141 is merged. You can try out on the master branch. I will move this PR to 2.7.0 first since we need to prepare the 2.6.0 release.

[lhotari]

PR #7141 is merged. You can try out on the master branch. I will move this PR to 2.7.0 first since we need to prepare the 2.6.0 release.

@codelipenghui PR #7141 doesn't seem to help with Pulsar Readers since Readers are always using exclusive subscriptions under the covers (

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

Line 48 in 208af7c

consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);

). Since a Pulsar Reader wraps a Consumer internally, I was thinking of the possibility of using a NonDurable Consumer directly without the Reader "wrapper". However this approach fails because creating NonDurable Consumers remains broken, that I have reported as #7436.

@codelipenghui would it be possible to extend the solution that was made in #7141 to also cover Readers?
I'm having issues with using seekAsync + hasMessageAvailableAsync on Readers. It seems that there are race conditions because of the current solution that disconnects from the broker side after the seek. The usage of seek + hasMessageAvailableAsync in Readers isn't deterministic at all. There's a separate issue about the issues with hasMessageAvailabe after seek, that's #7796 .