[fix][client] Java Client's Seek Logic Not Threadsafe #1

[JooHyukKim]

Fixes #19671

Motivation

To resolve the thread-safety issue around Java client's seek logic identified by #19671.

Modifications

• Made ConsumerImpl immediately log, cancel future and then return if duringSeek flag is not false.
  • by canceling the seekFuture, the client ensures that the current operation does not interfere with the ongoing seek operation and avoids any errors or inconsistencies that might otherwise occur.

Verifying this change

• Make sure that the change passes the CI checks.

I am not sure if my PR went through the intended CI checks...?

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

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: https://github.com/JooHyukKim/pulsar/pull/1

[nodece]

Could you add a test for your changes?

[JooHyukKim]

Could you add a test for your changes?

Sure! Will do in bit.

Can I ask for your opinion, @nodece ?

Do you think we should...

1. unit-test(mock everything)?
2. initialize everything?
3. or use reflection to extract the private method and test in isolation?

Because since the change is made in a private method I am trying to test through either of two methods,

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

Line 2168 in e0b50c9

public CompletableFuture<Void> seekAsync(long timestamp) {

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

Line 2178 in e0b50c9

public CompletableFuture<Void> seekAsync(MessageId originalMessageId) {

but I have not succeeded in finding a test I can refer to except below test from broker seems to cover too much.

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java

Line 185 in 9cb0503

public void testSeekToStart() throws Exception {

[tisonkun]

Shall we set:

        MessageIdAdv originSeekMessageId = seekMessageId;
        seekMessageId = (MessageIdAdv) seekId;

after the check passed?

[tisonkun]

@JooHyukKim BTW, edit comment won't send notification, I suggest you create new comment when you'd like to notify reviewers.

[tisonkun]

I'd appreciate it if you can provide a timing diagram to prove the fix. You may refer to apache/curator#430 as a typical description.

[nodece]

How to verify your changes?

You can set true for the duringSeek flag, and then call the seek method, you got a canceled future.

[JooHyukKim]

@JooHyukKim BTW, edit comment won't send notification, I suggest you create new comment when you'd like to notify reviewers.

Right, thank you for the reminder 👍🏻

[JooHyukKim]

I'd appreciate it if you can provide a timing diagram to prove the fix. You may refer to apache/curator#430 as a typical description.

Np 👍🏻. May I ask what tool everyone conventionally uses? Or your suggestion @tisonkun ?

[tisonkun]

tool

Generally, I just write in text :D

It's about:

1. Define the actors.
2. Describe at T_N, what actor takes what action
3. The order of T_N
4. What actions cause race condition

But any tool you searched is acceptable as long as the diagram is clear (with text, reviewers manually dump to diagram).

[tisonkun]

Wait a bit for review from @michaeljmarshall.

This patch looks good to me.

[JooHyukKim]

But any tool you searched is acceptable as long as the diagram is clear (with text, reviewers manually dump to diagram).

Thank you for the tips!

[tisonkun]

@JooHyukKim Please fix style violation by yourself. You can use CI workflow in your personal fork to verify. Or the command should be:

mvn -B -T 8 -ntp initialize apache-rat:check license:check
mvn -B -T 8 -ntp initialize checkstyle:check

Ping me once CI passed in your personal fork.

[JooHyukKim]

/pulsarbot rerun-failure-checks

[JooHyukKim]

Thank you @tisonkun, I made below two check style commands pass, "locally".

My personal fork is not running CI. Is it supposed to run though? 🤔🤔

mvn -B -T 8 -ntp initialize apache-rat:check license:check
mvn -B -T 8 -ntp initialize checkstyle:check

Ping me once CI passed in your personal fork.

[tisonkun]

personal fork is not running CI

Yes.

You may take lhotari#145 as an example. IIRC GitHub sometimes stuck on turning on Actions..I don't have too much suggestions here but I used to follow the instructions in Actions tab to add a sample workflow and then GitHub turn on Actions.

[JooHyukKim]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

Merging #20242 (4075a39) into master (9841364) will decrease coverage by 0.02%.
The diff coverage is 51.42%.

@@             Coverage Diff              @@
##             master   #20242      +/-   ##
============================================
- Coverage     72.87%   72.86%   -0.02%     
+ Complexity    31943     3601   -28342     
============================================
  Files          1868     1868              
  Lines        138594   138597       +3     
  Branches      15246    15247       +1     
============================================
- Hits         101003   100991      -12     
- Misses        29547    29557      +10     
- Partials       8044     8049       +5     

Flag	Coverage Δ
inttests	24.21% <22.85%> (-0.05%)	⬇️
systests	24.78% <20.00%> (-0.15%)	⬇️
unittests	72.18% <51.42%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	76.40% <51.42%> (-0.03%)	⬇️

... and 64 files with indirect coverage changes

[tisonkun]

Merging...

Thanks for your contribution!

[tisonkun]

@JooHyukKim May I ask what #1 means? It looks to me as part one or first attempt that I may wonder if you have outstanding works on this topic.

[JooHyukKim]

@JooHyukKim May I ask what #1 means?

Oh... 😅 "#1" means My first ever contribution to Pulsar!

I can remove the "#1" part from title if you think it will confuse the maintenance team. Just let me know~ 🙏🏼 @tisonkun

Thanks for your contribution!

@tisonkun Thank you for great guidance on your part 🙏🏼

[tisonkun]

Never mind. I can edit the commit message before merging :D

[michaeljmarshall]

Thanks for your contribution @JooHyukKim! Would you mind updating the Javadoc for the seek and seekAsync methods to indicate what will happen if a user calls a seek method before a previous call completes? Here is a reference to one of the Javadocs:

pulsar/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java

Lines 529 to 537 in a918651

	/**
	* Reset the subscription associated with this consumer to a specific message publish time.
	*
	* @param timestamp
	* the message publish time where to reposition the subscription
	* The timestamp format should be Unix time in milliseconds.
	* @return a future to track the completion of the seek operation
	*/
	CompletableFuture<Void> seekAsync(long timestamp);

[JooHyukKim]

Thanks for your contribution @JooHyukKim! Would you mind updating the Javadoc for the seek and seekAsync methods to indicate what will happen if a user calls a seek method before a previous call completes?

Will do! 🙏🏽 Later after work 😀

[Technoboy-]

@tisonkun
For this patch, maybe you forget to set the milestone. and there are some modifications not related to the issue

[tisonkun]

@Technoboy- Thanks for setting up the milestone.

not related to the issue

Yep. I'm investigating some tools to unify whitespace so that we don't ask for aligning again and again.