fix lastCumulativeAck.messageId npe when PersistentAcknowledgmentsGroupingTracker.flushSync #12343

gaozhangmin · 2021-10-13T11:42:26Z

Reader ACK would be stuck:

This was caused by:
flushAsync is executed by scheduledTask, Any thrown exception or error reaching the executor causes the executor to halt。

The exception is as below:

gaozhangmin · 2021-10-13T11:50:24Z

@BewareMyPower @eolivelli @codelipenghui PTAL

gaozhangmin · 2021-10-14T03:43:32Z

/pulsarbot run-failure-checks

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

315157973 · 2021-11-01T08:21:16Z

@eolivelli PTAL agagin

gaozhangmin · 2021-11-18T02:57:39Z

@eolivelli PTAL

…upingTracker.flushSync

gaozhangmin · 2021-11-19T05:47:26Z

/pulsarbot run-failure-checks

gaozhangmin · 2021-11-19T08:51:59Z

/pulsarbot run-failure-checks

codelipenghui · 2021-11-21T03:30:41Z

@eolivelli Could you please help review this PR again?

github-actions · 2021-11-21T03:31:13Z

@gaozhangmin:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

michaeljmarshall · 2021-11-21T21:32:38Z

Regarding the halted executor, I believe @lhotari is adding support to ensure that these exceptions would not stop the scheduled task.

michaeljmarshall

I'm wondering how the client gets into a state where we have cumulativeAckFlushRequired set to true, but there is no "last ack". Is there a chance that we're updating the state incorrectly?

michaeljmarshall · 2021-11-21T21:49:11Z

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

-            this.consumer.unAckedChunkedMessageIdSequenceMap.remove(lastCumulativeAck.messageId);
-            shouldFlush = true;
-            cumulativeAckFlushRequired = false;
+            final MessageIdImpl messageIdOfLastAck = lastCumulativeAck.messageId;


Can you explain why we enter this code block when lastCumulativeAck.messageId is null? It seems like cumulativeAckFlushRequired should be false. I'm fine with adding a null check, but it seems like there could be another issue.

The previous logic showed, stats were updated only when this.consumer.unAckedChunkedMessageIdSequenceMap.remove successfully.

@gaozhangmin When reading the code, it seems that messageId should never be null. As @michaeljmarshall pointed out, there could be another issue hiding here.
It might be a thread safety issue which could cause other problems. @gaozhangmin Did you check how messageId could become null?

@lhotari No, i didn't check the reason. much same as this issue: #11607

PR #11607 fixes another case where lastCumulativeAck.messageId was null. @BewareMyPower What are your thoughts about the reason why the field is null in the first place? Could there be a thread safety issue that won't go away by adding null checks?

@gaozhangmin do you have a chance to analyse the reason? My concern is that adding a null check might just silently ignore a potential thread safety issue which might lead to inconsistency behavior.

@lhotari As I've explained in my PR, I think there is a possible thread safety problem like

if (lastCumulativeAck.messageId == null) { // 1. messageId is not null return false; } // 2. messageId was modified to null in another thread if (messageId.compareTo(lastCumulativeAck.messageId) <= 0) { // 3. messageId is null now

I think the root cause is the design of LastCumulativeAck class. It's not well encapsulated. Though its fields are all private, the class is an inner class so the outer class can access the members directly. And we can see the direct access in many places, which makes it hard to analyze whether all the accesses are thread safe.

I think the root cause is the design of LastCumulativeAck class

@BewareMyPower I also think that LastCumulativeAck class is the source of problems. Using the Netty Recycler introduces the thread safety issue. Using an immutable class design for LastCumulativeAck and removing the use of Netty recycler would be something that I'd recommend for fixing the issue.

Sure, I will implement immutable class design.
What's about this pr. should we close it? @lhotari

@gaozhangmin @lhotari I picked up this issue today. After looking deeper into this issue, I found another bug and it would be better to synchronize the update operations of LastCumulativeAck. See #16072, PTAL when you have time.

lhotari · 2021-11-22T07:16:49Z

Regarding the halted executor, I believe @lhotari is adding support to ensure that these exceptions would not stop the scheduled task.

@michaeljmarshall The PR #12853 has already been merged to master.

gaozhangmin · 2021-11-23T02:19:39Z

/pulsarbot run-failure-checks

Anonymitaet · 2021-11-24T08:45:54Z

@gaozhangmin is this a bug fix (no need to update docs)?

github-actions · 2021-11-24T08:52:48Z

@gaozhangmin:Thanks for providing doc info!

gaozhangmin · 2021-11-24T09:03:41Z

@gaozhangmin is this a bug fix (no need to update docs)?

Yes

Anonymitaet · 2021-11-24T10:03:57Z

@gaozhangmin when submitting a PR, can you help provide a doc label (tick the box) in the PR template which contains info about doc? This helps others know more about the changes. Thanks

gaozhangmin · 2021-11-26T09:26:32Z

/pulsarbot run-failure-checks

eolivelli · 2021-11-27T07:55:32Z

Yes I would close this pr

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, two synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values. - `moveOwnershipTo`: This method moves the ownership to another `LastCumulativeAck` object, which will be responsible to recycle the `BitSetRecyclable` field after that. With the methods above, each time `flushAsync` is called, move the ownership of `lastCumulativeAck` field to another thread local field to send the ACK command and recycle the `BitSetRecyclable` field. - `lastCumulativeAck` updates the latest message ID and bit set, the update operations can be performed by multiple threads and `lastCumulativeAck` saves the latest value. - `threadLocalLastCumulativeAckToFlush` only acts as a temporary cache to the latest value in `flushAsync`.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values. - `moveOwnershipTo`: This method moves the ownership to another `LastCumulativeAck` object. After that, the `update` operation on this object won't recycle the `BitSetRecyclable` field. - `restoreOwnershipIfEmpty`: Restore the ownership from another `LastCumulativeAck` object. With the methods above, each time `flushAsync` is called, move the ownership of `lastCumulativeAck` field to another thread local field to send the ACK command. After that, restore the ownership to `lastCumulativeAck` unless it has been updated in other threads.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. Then mark it as no need to flush. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. Remove unused field Don't reset in LastCumulativeAck#flush

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice.

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd)

…ache#16072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd) (cherry picked from commit 5eefdf1)

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd)

BewareMyPower assigned gaozhangmin Oct 14, 2021

BewareMyPower added release/2.8.2 type/bug The PR fixed a bug or issue reported a bug labels Oct 14, 2021

BewareMyPower requested review from codelipenghui, congbobo184 and eolivelli October 14, 2021 07:44

BewareMyPower reviewed Oct 14, 2021

View reviewed changes

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java Outdated Show resolved Hide resolved

eolivelli requested changes Oct 14, 2021

View reviewed changes

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java Outdated Show resolved Hide resolved

gaozhangmin force-pushed the fix-npe-lastCumulativeAck-messageId branch from a963849 to e65e64d Compare October 14, 2021 09:35

gaozhangmin requested review from BewareMyPower and eolivelli October 14, 2021 09:37

BewareMyPower approved these changes Oct 14, 2021

View reviewed changes

gaozhangmin force-pushed the fix-npe-lastCumulativeAck-messageId branch 3 times, most recently from aa42602 to b13d274 Compare October 21, 2021 06:52

syhily mentioned this pull request Nov 4, 2021

Check the topic's markDeletePosition and readPosition are very different streamnative/pulsar-flink#438

Closed

codelipenghui approved these changes Nov 10, 2021

View reviewed changes

gaozhangmin mentioned this pull request Nov 17, 2021

[bug](indeviduallyDeleteMessage): slow rate consumer but ack in order will get indeviduallyDeleteMessage #12759

Closed

fix lastCumulativeAck.messageId npe when PersistentAcknowledgmentsGro…

06d41bb

…upingTracker.flushSync

gaozhangmin force-pushed the fix-npe-lastCumulativeAck-messageId branch from b13d274 to 06d41bb Compare November 19, 2021 02:21

315157973 approved these changes Nov 20, 2021

View reviewed changes

codelipenghui added this to the 2.10.0 milestone Nov 21, 2021

codelipenghui removed the release/2.8.2 label Nov 21, 2021

codelipenghui added the release/2.8.3 label Nov 21, 2021

github-actions bot added the doc-label-missing label Nov 21, 2021

eolivelli approved these changes Nov 21, 2021

View reviewed changes

michaeljmarshall reviewed Nov 21, 2021

View reviewed changes

lhotari added the doc-not-needed Your PR changes do not impact docs label Nov 24, 2021

github-actions bot removed the doc-label-missing label Nov 24, 2021

gaozhangmin closed this Dec 3, 2021

zymap removed the release/2.8.3 label Dec 23, 2021

BewareMyPower mentioned this pull request Jun 15, 2022

[fix][Java Client] Fix thread safety issue of LastCumulativeAck #16072

Merged

4 tasks

fix lastCumulativeAck.messageId npe when PersistentAcknowledgmentsGroupingTracker.flushSync #12343

fix lastCumulativeAck.messageId npe when PersistentAcknowledgmentsGroupingTracker.flushSync #12343

Uh oh!

Conversation

gaozhangmin commented Oct 13, 2021

Uh oh!

gaozhangmin commented Oct 13, 2021

Uh oh!

gaozhangmin commented Oct 14, 2021

Uh oh!

Uh oh!

Uh oh!

315157973 commented Nov 1, 2021

Uh oh!

gaozhangmin commented Nov 18, 2021

Uh oh!

gaozhangmin commented Nov 19, 2021

Uh oh!

gaozhangmin commented Nov 19, 2021

Uh oh!

codelipenghui commented Nov 21, 2021

Uh oh!

github-actions bot commented Nov 21, 2021

Uh oh!

michaeljmarshall commented Nov 21, 2021

Uh oh!

michaeljmarshall left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gaozhangmin Nov 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lhotari Nov 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lhotari commented Nov 22, 2021

Uh oh!

gaozhangmin commented Nov 23, 2021

Uh oh!

Anonymitaet commented Nov 24, 2021

Uh oh!

github-actions bot commented Nov 24, 2021

Uh oh!

gaozhangmin commented Nov 24, 2021

Uh oh!

Anonymitaet commented Nov 24, 2021

Uh oh!

gaozhangmin commented Nov 26, 2021

Uh oh!

eolivelli commented Nov 27, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

gaozhangmin Nov 23, 2021 •

edited

Loading

lhotari Nov 23, 2021 •

edited

Loading