[fix] [tx] Transaction buffer recover blocked by readNext by poorbarcode · Pull Request #18833 · apache/pulsar

poorbarcode · 2022-12-09T08:54:15Z

Motivation

Context for Transaction Buffer

If turn on transactionCoordinatorEnabled, then TransactionBuffer will be initialized when a user topic create.
The TransactionBuffer reads the aborted logs of transactions from topic __transaction_buffer_snapshot -- this process is called recovery.
During recovery, the reading from that snapshot ledger is done via a Reader; the reader works like this:

while (reader.hasMessageAvailable()){
    reader.readNext();
}

Context for Compaction

After pip-14, the consumer that enabled feature read-compacted will read messages from the compacted topic instead of the original topic if the task-compaction is done, and read messages from the original topic if task-compaction is not done.
If the data of the last message with key k sent to a topic is null, the compactor will mark all messages for that key as deleted.

Issue

There is a race condition: after executing reader.hasMessageAvailable, the following messages have been deleted by compaction-task, so read next will be blocked because there have no messages to read.

Modifications

If hits this issue, do recover again.

Why not just let the client try to load the topic again to retry the recover?

If the topic load is failed, the client will receive an error response. This is a behavior that we can handle, so should not be perceived by the users.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[ci] [fix] [tx] Transaction buffer recover can not completed poorbarcode/pulsar#48

gaoran10 · 2022-12-09T08:57:33Z

                        }
-                        closeReader(reader);
                        return CompletableFuture.completedFuture(startReadCursorPosition);
+                    } catch (NullPointerException npe) {


Could we check if the message is null or not directly?

Already fixed, thanks

poorbarcode · 2022-12-09T09:16:37Z

This PR should cherry-pick into these branches:

branch-2.9 @congbobo184
branch-2.10
branch-2.11 @liangyepianzhou

BewareMyPower · 2022-12-09T10:36:39Z

                    try {
                        while (reader.hasMoreEvents()) {
-                            Message<TransactionBufferSnapshot> message = reader.readNext();
+                            Message<TransactionBufferSnapshot> message = reader.readNext(2, TimeUnit.SECONDS);


It treats a message that cannot be received in 2 seconds as an exceptional case that the message has been deleted. I think it should be done in SystemTopicClient#readNext instead of adding another public API. If there are other places (e.g. a protocol handler that could access this interface) that calls SystemTopicClient#readNext without the timeout, the error should be processed in the same way.

I'd rather implement the wrapped readNext() like:

public Message<T> readNext() throws PulsarClientException { Message<T> message = reader.readNext(2, TimeUnit.SECONDS); if (message == null) { throw new PulsarClientException.InvalidMessageException( "When reading from topic " + reader.getTopic() + ", the latest message has been deleted"); } return message; }

Then the exception could be handled in the catch case:

try { while (reader.hasMoreEvents()) { Message<TransactionBufferSnapshot> message = reader.readNext(); /* ... */ } catch (PulsarClientException.InvalidMessageException e) { return FutureUtil.failedFuture( new TransactionBufferException.TBRecoverCantCompletedException(e.getMessage()));

In addition, I'm wondering if the 2 seconds is reasonable. What if the message actually exists and the timeout happened only because of the network issue?

I think it should be done in SystemTopicClient#readNext instead of adding another public API. If there are other places (e.g. a protocol handler that could access this interface) that calls SystemTopicClient#readNext without the timeout, the error should be processed in the same way.

I think it is a good suggestion. This also leads to a lack of flexibility in the API. For example, if there is no message at the moment and I need to wait for the following message to be sent, there is no API to use. Since the 'SystemTopicClient.Reader' is an internal-only API, I feel it would be better to provide better usability.

In addition, I'm wondering if the 2 seconds is reasonable. What if the message actually exists and the timeout happened only because of the network issue?

Yes, you are right. we can't identify what is happening.

The current fix is like a patch to fix this existing issue, and there will be one discussion to fix the root cause:

When the client already gets the last message id and caches it, then the compaction task will delete the messages end of this topic, do change the logic of the compaction task: the last message of this topic will be retained

Hi @BewareMyPower

I used the simper logic suggested by @congbobo184, could you retake a look?

congbobo184 · 2022-12-12T02:38:13Z

                    try {
                        while (reader.hasMoreEvents()) {
-                            Message<TransactionBufferSnapshot> message = reader.readNext();
+                            Message<TransactionBufferSnapshot> message = reader.readNext(2, TimeUnit.SECONDS);


try { Message<TransactionBufferSnapshot> message = reader.readNextAsync().get(30, TimeUnit.SECONDS); } catch (Exception ex) { Throwable t = FutureUtil.unwrapCompletionException(ex); log.error("[{}] Transaction buffer recover fail when read " + "transactionBufferSnapshot!", topic.getName(), t); closeReader(reader); return FutureUtil.failedFuture(t); }

better to use this logic, topicTransactionBuffer use getTransactionExecutorProvider thread recover, so we don't need to care about the deadlock problem. and don't need to add any new API.
2. it's better to use timeout client operationTimeoutMs, 2 seconds is too short

Good suggestion, already fixed

How to try recovery again if implemented in this way?

IMO, we should add a TimeoutException check here.

IMO, we should add a TimeoutException check here.

Yes, you are correct. already fixed

How to try recovery again if implemented in this way?

The next recover will be triggered by the client.

liangyepianzhou · 2022-12-12T03:01:45Z

+                                                + " been deleted by compaction-task or trim ledger.",
+                                        topic.getName(),
+                                        SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT);
+                                log.warn(warnLog);


Maybe there should use the debug log level because this is a normal case.

Hi @liangyepianzhou

I used the simper logic suggested by @congbobo184, could you retake a look?

congbobo184 · 2022-12-12T09:31:41Z

-                                return FutureUtil.failedFuture(
-                                        new TransactionBufferException.TBRecoverCantCompletedException(warnLog));
+                            Message<TransactionBufferSnapshot> message;
+                            try {


how about moving this try block on the line-89 try block?

already fixed

congbobo184 · 2022-12-12T12:03:36Z

+            assertEquals(compaction.getMarkDeletedPosition().getLedgerId(),
+                    managedLedger.getLastConfirmedEntry().getLedgerId());
+            assertEquals(compaction.getMarkDeletedPosition().getEntryId(),


Suggested change

assertEquals(compaction.getMarkDeletedPosition().getLedgerId(),

managedLedger.getLastConfirmedEntry().getLedgerId());

assertEquals(compaction.getMarkDeletedPosition().getEntryId(),

assertEquals(compaction.getMarkDeletedPosition(), managedLedger.getLastConfirmedEntry());

Already fixed

congbobo184 · 2022-12-12T12:17:33Z

+                            try {
+                                message = reader.readNextAsync().get(30, TimeUnit.SECONDS);
+                            } catch (Exception ex) {
+                                Throwable t = FutureUtil.unwrapCompletionException(ex);


please wrap TimeoutException, otherwise, tc will not retry the end tb op and client will not reconnect

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

Lines 182 to 187 in 1be5a69

if (e instanceof PulsarClientException) {

// if transaction buffer recover fail throw PulsarClientException,

// we need to change the PulsarClientException to ServiceUnitNotReadyException,

// the tc do op will retry

transactionBufferFuture.completeExceptionally

(new BrokerServiceException.ServiceUnitNotReadyException(e.getMessage(), e));

please check the return error code, current return error code seem not correct.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Lines 1405 to 1412 in 1be5a69

Throwable cause = exception.getCause();

log.error("producerId {}, requestId {} : TransactionBuffer recover failed",

producerId, requestId, exception);

if (producerFuture.completeExceptionally(exception)) {

commandSender.sendErrorResponse(requestId,

ServiceUnitNotReadyException.getClientErrorCode(cause),

cause.getMessage());

}

Already fixed.

…Data

codecov-commenter · 2022-12-16T02:02:43Z

Codecov Report

❌ Patch coverage is 50.00000% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.61%. Comparing base (3180a4a) to head (01d5cec).
⚠️ Report is 3156 commits behind head on master.

Files with missing lines	Patch %	Lines
...er/impl/SingleSnapshotAbortedTxnProcessorImpl.java	50.00%	6 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18833      +/-   ##
============================================
+ Coverage     46.17%   46.61%   +0.44%     
- Complexity    10359    10472     +113     
============================================
  Files           703      703              
  Lines         68845    68866      +21     
  Branches       7382     7383       +1     
============================================
+ Hits          31788    32103     +315     
+ Misses        33448    33145     -303     
- Partials       3609     3618       +9

Flag	Coverage Δ
unittests	`46.61% <50.00%> (+0.44%)`	⬆️

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

Files with missing lines	Coverage Δ
...er/impl/SingleSnapshotAbortedTxnProcessorImpl.java	`70.65% <50.00%> (-3.16%)`	⬇️

... and 61 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…xt (#18971) ### Motivation Since PR #18833 can not cherry-pick to `branch-2.10`, create a separate PR. #### Context for Transaction Buffer - If turn on `transactionCoordinatorEnabled`, then `TransactionBuffer` will be initialized when a user topic create. - The `TransactionBuffer` reads the aborted logs of transactions from topic `__transaction_buffer_snapshot` -- this process is called `recovery`. - During recovery, the reading from that snapshot ledger is done via a `Reader`; the reader works like this: ``` while (reader.hasMessageAvailable()){ reader.readNext(); } ``` #### Context for Compaction - After [pip-14](https://github.com/apache/pulsar/wiki/PIP-14:-Topic-compaction), the consumer that enabled feature read-compacted will read messages from the compacted topic instead of the original topic if the task-compaction is done, and read messages from the original topic if task-compaction is not done. - If the data of the last message with key k sent to a topic is null, the compactor will mark all messages for that key as deleted. #### Issue There is a race condition: after executing `reader.hasMessageAvailable`, the following messages have been deleted by compaction-task, so read next will be blocked because there have no messages to read. ---- ### Modifications - If hits this issue, do recover again. ---- #### Why not just let the client try to load the topic again to retry the recover? If the topic load is failed, the client will receive an error response. This is a behavior that we can handle, so should not be perceived by the users.

#18970) ### Motivation Since PR #18833 can not cherry-pick to `branch-2.9`, create a separate PR. #### Context for Transaction Buffer - If turn on `transactionCoordinatorEnabled`, then `TransactionBuffer` will be initialized when a user topic create. - The `TransactionBuffer` reads the aborted logs of transactions from topic `__transaction_buffer_snapshot` -- this process is called `recovery`. - During recovery, the reading from that snapshot ledger is done via a `Reader`; the reader works like this: ``` while (reader.hasMessageAvailable()){ reader.readNext(); } ``` #### Context for Compaction - After [pip-14](https://github.com/apache/pulsar/wiki/PIP-14:-Topic-compaction), the consumer that enabled feature read-compacted will read messages from the compacted topic instead of the original topic if the task-compaction is done, and read messages from the original topic if task-compaction is not done. - If the data of the last message with key k sent to a topic is null, the compactor will mark all messages for that key as deleted. #### Issue There is a race condition: after executing `reader.hasMessageAvailable`, the following messages have been deleted by compaction-task, so read next will be blocked because there have no messages to read. ---- ### Modifications - If hits this issue, do recover again. ---- #### Why not just let the client try to load the topic again to retry the recover? If the topic load is failed, the client will receive an error response. This is a behavior that we can handle, so should not be perceived by the users.

…xt (#18969) ### Motivation Since PR #18833 can not cherry-pick to `branch-2.11`, create a separate PR. #### Context for Transaction Buffer - If turn on `transactionCoordinatorEnabled`, then `TransactionBuffer` will be initialized when a user topic create. - The `TransactionBuffer` reads the aborted logs of transactions from topic `__transaction_buffer_snapshot` -- this process is called `recovery`. - During recovery, the reading from that snapshot ledger is done via a `Reader`; the reader works like this: ``` while (reader.hasMessageAvailable()){ reader.readNext(); } ``` #### Context for Compaction - After [pip-14](https://github.com/apache/pulsar/wiki/PIP-14:-Topic-compaction), the consumer that enabled feature read-compacted will read messages from the compacted topic instead of the original topic if the task-compaction is done, and read messages from the original topic if task-compaction is not done. - If the data of the last message with key k sent to a topic is null, the compactor will mark all messages for that key as deleted. #### Issue There is a race condition: after executing `reader.hasMessageAvailable`, the following messages have been deleted by compaction-task, so read next will be blocked because there have no messages to read. ---- ### Modifications - If hits this issue, do recover again. ---- #### Why not just let the client try to load the topic again to retry the recover? If the topic load is failed, the client will receive an error response. This is a behavior that we can handle, so should not be perceived by the users.

…xt (apache#18971) ### Motivation Since PR apache#18833 can not cherry-pick to `branch-2.10`, create a separate PR. #### Context for Transaction Buffer - If turn on `transactionCoordinatorEnabled`, then `TransactionBuffer` will be initialized when a user topic create. - The `TransactionBuffer` reads the aborted logs of transactions from topic `__transaction_buffer_snapshot` -- this process is called `recovery`. - During recovery, the reading from that snapshot ledger is done via a `Reader`; the reader works like this: ``` while (reader.hasMessageAvailable()){ reader.readNext(); } ``` #### Context for Compaction - After [pip-14](https://github.com/apache/pulsar/wiki/PIP-14:-Topic-compaction), the consumer that enabled feature read-compacted will read messages from the compacted topic instead of the original topic if the task-compaction is done, and read messages from the original topic if task-compaction is not done. - If the data of the last message with key k sent to a topic is null, the compactor will mark all messages for that key as deleted. #### Issue There is a race condition: after executing `reader.hasMessageAvailable`, the following messages have been deleted by compaction-task, so read next will be blocked because there have no messages to read. ---- ### Modifications - If hits this issue, do recover again. ---- #### Why not just let the client try to load the topic again to retry the recover? If the topic load is failed, the client will receive an error response. This is a behavior that we can handle, so should not be perceived by the users. (cherry picked from commit f7dc64f)

Technoboy- · 2023-02-08T08:00:59Z

Introduced by #17847, no need to release for other branch

3286360470 · 2023-04-10T02:23:08Z

Are there steps to reproduce this issue stably? I have restarted the broker many times locally and still haven't reproduced the snapshot stuck situation.

congbobo184 · 2023-04-10T04:20:27Z

@3286360470 This reproduction requires steps to delete topics, otherwise it cannot be reproduced

github-actions Bot added the doc-not-needed Your PR changes do not impact docs label Dec 9, 2022

poorbarcode changed the title ~~[fix] [tx] Transaction buffer recover can not completed~~ [fix] [tx] Transaction buffer recover blocked by readNext Dec 9, 2022

gaoran10 reviewed Dec 9, 2022

View reviewed changes

BewareMyPower requested changes Dec 9, 2022

View reviewed changes

Comment thread ...er/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java

BewareMyPower requested changes Dec 9, 2022

View reviewed changes

poorbarcode force-pushed the reproduce/tb_trim_ledger_4 branch from 99f1757 to 1697752 Compare December 11, 2022 13:19

poorbarcode requested review from BewareMyPower and gaoran10 and removed request for BewareMyPower December 12, 2022 00:38

congbobo184 reviewed Dec 12, 2022

View reviewed changes

liangyepianzhou assigned poorbarcode Dec 12, 2022

liangyepianzhou reviewed Dec 12, 2022

View reviewed changes

congbobo184 reviewed Dec 12, 2022

View reviewed changes

poorbarcode added 7 commits December 14, 2022 15:56

[improve] [test] Add test testTrimLedgerWillKeepsAtLeastOneLedgerWith…

fc32b1f

…Data

[fix] [tx] Transaction buffer recover can not completed

ac8c548

hanle npe more precision

0dd2818

avoid unnecessary changes

cdff5e1

Simplified logic

73f90ef

fix test

1718782

address comments

51d1f64

poorbarcode force-pushed the reproduce/tb_trim_ledger_4 branch from 20ff97e to 51d1f64 Compare December 14, 2022 07:56

poorbarcode requested review from BewareMyPower, congbobo184, gaoran10 and liangyepianzhou and removed request for BewareMyPower, congbobo184, gaoran10 and liangyepianzhou December 14, 2022 08:00

make test simpler

01d5cec

codelipenghui approved these changes Dec 16, 2022

View reviewed changes

codelipenghui added this to the 2.12.0 milestone Dec 16, 2022

codelipenghui added type/bug The PR fixed a bug or issue reported a bug area/broker release/2.11.1 release/2.9.5 release/2.10.4 labels Dec 16, 2022

codelipenghui merged commit b9446aa into apache:master Dec 17, 2022

This was referenced Dec 17, 2022

[fix] [tx] [branch-2.11] Transaction buffer recover blocked by readNext #18969

Merged

[fix] [tx] [branch-2.9] Transaction buffer recover blocked by readNext #18970

Merged

[fix] [tx] [branch-2.10] Transaction buffer recover blocked by readNext #18971

Merged

poorbarcode deleted the reproduce/tb_trim_ledger_4 branch December 19, 2022 01:56

poorbarcode mentioned this pull request Dec 28, 2022

[feat][txn] implement the SnapshotSegmentAbortedTxnProcessor #18273

Merged

1 task

liangyepianzhou added release/2.10.3 and removed release/2.10.4 labels Jan 4, 2023

liangyepianzhou added release/2.10.4 and removed release/2.10.3 labels Feb 1, 2023

Technoboy- removed release/2.11.1 release/2.9.5 release/2.10.4 labels Feb 8, 2023

	if (e instanceof PulsarClientException) {
	// if transaction buffer recover fail throw PulsarClientException,
	// we need to change the PulsarClientException to ServiceUnitNotReadyException,
	// the tc do op will retry
	transactionBufferFuture.completeExceptionally
	(new BrokerServiceException.ServiceUnitNotReadyException(e.getMessage(), e));

	Throwable cause = exception.getCause();
	log.error("producerId {}, requestId {} : TransactionBuffer recover failed",
	producerId, requestId, exception);
	if (producerFuture.completeExceptionally(exception)) {
	commandSender.sendErrorResponse(requestId,
	ServiceUnitNotReadyException.getClientErrorCode(cause),
	cause.getMessage());
	}

Uh oh!

Conversation

poorbarcode commented Dec 9, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Context for Transaction Buffer

Context for Compaction

Issue

Modifications

Why not just let the client try to load the topic again to retry the recover?

Documentation

Matching PR in forked repository

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode commented Dec 9, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

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!

poorbarcode Dec 10, 2022 • 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!

poorbarcode Dec 12, 2022 • 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!

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!

Choose a reason for hiding this comment

Uh oh!

congbobo184 Dec 12, 2022 • 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!

codecov-commenter commented Dec 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Technoboy- commented Feb 8, 2023

Uh oh!

3286360470 commented Apr 10, 2023

Uh oh!

congbobo184 commented Apr 10, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

poorbarcode commented Dec 9, 2022 •

edited

Loading

poorbarcode commented Dec 9, 2022 •

edited

Loading

poorbarcode Dec 10, 2022 •

edited

Loading

poorbarcode Dec 12, 2022 •

edited

Loading

congbobo184 Dec 12, 2022 •

edited

Loading

codecov-commenter commented Dec 16, 2022 •

edited

Loading