[fix][txn] fix NPE of TransactionMetaStoreHandler

[congbobo184]

05:30:15.799 [pulsar-client-io-14-1] ERROR org.apache.pulsar.client.impl.ProducerImpl - [persistent://public/default/sink-topic-partition-7] [normal-pulsar-1-58] Failed to create producer: Disconnected from server at 127.0.0.1/127.0.0.1:6650
05:30:15.799 [pulsar-client-internal-19-1] ERROR org.apache.pulsar.client.impl.TransactionMetaStoreHandler - The cnx was null when the TC handler was ready
java.lang.NullPointerException: null
  at org.apache.pulsar.client.impl.TransactionMetaStoreHandler.checkStateAndSendRequest(TransactionMetaStoreHandler.java:664) ~[test-transaction-1.0-SNAPSHOT-jar-with-dependencies.jar:?]
  at org.apache.pulsar.client.impl.TransactionMetaStoreHandler.lambda$newTransactionAsync$7(TransactionMetaStoreHandler.java:206) ~[test-transaction-1.0-SNAPSHOT-jar-with-dependencies.jar:?]
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[?:?]
  at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
  at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[test-transaction-1.0-SNAPSHOT-jar-with-dependencies.jar:?]
  at java.lang.Thread.run(Thread.java:829) ~[?:?]

Motivation

fix transactionMetadataStoreHandle cnx is null.

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

Lines 134 to 154 in b13d15c

	connectionHandler.setClientCnx(cnx);
	cnx.registerTransactionMetaStoreHandler(transactionCoordinatorId, this);
	// if broker protocol version < 19, don't send TcClientConnectRequest to broker.
	if (cnx.getRemoteEndpointProtocolVersion() > ProtocolVersion.v18.getValue()) {
	long requestId = client.newRequestId();
	ByteBuf request = Commands.newTcClientConnectRequest(transactionCoordinatorId, requestId);
	cnx.sendRequestWithId(request, requestId).thenRun(() -> {
	internalPinnedExecutor.execute(() -> {
	LOG.info("Transaction coordinator client connect success! tcId : {}", transactionCoordinatorId);
	if (!changeToReadyState()) {
	setState(State.Closed);
	cnx.channel().close();
	}
	if (!this.connectFuture.isDone()) {
	this.connectFuture.complete(null);
	}
	this.connectionHandler.resetBackoff();
	pendingRequests.forEach((requestID, opBase) -> checkStateAndSendRequest(opBase));

because

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

Line 142 in b13d15c

cnx.sendRequestWithId(request, requestId).thenRun(() -> {

is the async method, set cnx, and change to ready state is not an atomic operation.

1. set cnx into this handle
2. reconnect will set the cnx to null

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

   Line 101 in 29308e4

   CLIENT_CNX_UPDATER.set(this, null);

3. then send connect request and get a response that will change state to ready state.
4. the client will not reconnect because the state is ready
5. the cnx also is null. transactionMetadataStoreHandle will not do txn op.

Modifications

fix race condition

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull requestintroducese a new feature? (yes)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create afollow-upp issue for adding the documentation

[HQebupt]

/pulsarbot run-failure-checks

[mattisonchao]

LGTM,