[fix] [txn] update txn status concurrently throw exception

[TakaHiR07]

Motivation

When MLTransactionMetadataStore#updateTxnStatus concurrently, there is a case only one request succeed, other requests throw exception.

Modifications

If the newStatus == txnMeta.status(), it should not throw exception to client. Which is the same as

pulsar/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStore.java

Lines 385 to 388 in 7ea8741

	if (txnMetaListPair.getLeft().status() == newStatus) {
	promise.complete(null);
	return promise;
	}

1. judge "txnMetaListPair.getLeft().status() == newStatus" again after synchronize
2. add test

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: TakaHiR07#4

[github-actions]

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

[TakaHiR07]

@congbobo184 can you help take a look of this problem?

[liangyepianzhou]

Good catch!

[poorbarcode]

@congbobo184 @liangyepianzhou

Is this PR similar to #19630?

[liangyepianzhou]

@congbobo184 @liangyepianzhou

Is this PR similar to #19630?

The modification is similar to #19630, but the motivation is different.
The motivation for #19630 is try to commit a transaction which has been committed already.
In this case, the modification is invalid because the TxnMeta of the transaction is already deleted.

[liangyepianzhou]

TxnStats is written by a single thread, so the update stats thread should also be single-threaded. I don't understand where the concurrency problem occurs. Can you point it out?

        CompletableFuture
                .runAsync(
                        () -> internalAsyncAddData(data, callback, ctx), singleThreadExecutorForWrite)
                .exceptionally(e -> {
                    log.warn("Execute 'internalAsyncAddData' fail", e);
                    return null;
                });

[TakaHiR07]

TxnStats is written by a single thread, so the update stats thread should also be single-threaded. I don't understand where the concurrency problem occurs. Can you point it out?

TxnStatus is appended to txnLog by single thread, but the operation after thenAccept is executed in other thread. So it cause the concurrency problem.

pulsar/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStore.java

Lines 398 to 400 in ac46e2e

	return transactionLog.append(transactionMetadataEntry)
	.thenAccept(position -> {
	appendLogCount.increment();

[liangyepianzhou]

TxnStatus is appended to txnLog by single thread, but the operation after thenAccept is executed in other thread. So it cause the concurrency problem.

The thenAccept will use the original thread instead of another thread, right?
Please take a look. @poorbarcode

[github-actions]

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