[fix][txn] Set TC client state update and mistake completed future exceptionally.

[liangyepianzhou]

Motivation

1. Fix the logic flow in startAsync.
   1. change the state of the TCClient to STARTING
   2. Connecting to the transaction coordinators
   3. Set the state to READY
   4. Connect fails but the state is READY
2. Fix the future completed logic flow.
   1. connect to the transaction coordinator.
   2. get a retryable exception
   3. complete the connected future with the exception
   4. do retry the operation
   5. complete the future again.

Modification

1. Update the state after all connections are completed or catch exceptions.
2. Do not complete the future if the exception is a retryable future.

Verifying this change

• Make sure that the change passes the 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).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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:

[eolivelli]

I have two questions/points:

1. what happens if some of the coordinators are not available ?

2. is this change adding more latency in the creation of the PulsarClient ?

[liangyepianzhou]

I have two questions/points:

1. what happens if some of the coordinators are not available ?
2. is this change adding more latency in the creation of the PulsarClient ?

The TC client is created synchronized, so this change will not add any latency.
If the coordinators are unavailable, the client will try to connect later.
This modification will not bring any effects on these two questions.

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

Lines 218 to 226 in ea4f7eb

	if (conf.isEnableTransaction()) {
	tcClient = new TransactionCoordinatorClientImpl(this);
	try {
	tcClient.start();
	} catch (Throwable e) {
	log.error("Start transactionCoordinatorClient error.", e);
	throw new PulsarClientException(e);
	}
	}

[congbobo184]

How the connect request can retry?

[liangyepianzhou]

It will do a retry operation after getting an exception.

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

Lines 150 to 180 in c17f355

	}).exceptionally((e) -> {
	internalPinnedExecutor.execute(() -> {
	LOG.error("Transaction coordinator client connect fail! tcId : {}",
	transactionCoordinatorId, e.getCause());
	if (getState() == State.Closing || getState() == State.Closed
	|| e.getCause() instanceof PulsarClientException.NotAllowedException) {
	setState(State.Closed);
	cnx.channel().close();
	} else {
	connectionHandler.reconnectLater(e.getCause());
	}
	});
	return null;
	});
	} else {
	registerToConnection(cnx);
	}
	});
	}
	private boolean registerToConnection(ClientCnx cnx) {
	if (changeToReadyState()) {
	connectionHandler.setClientCnx(cnx);
	cnx.registerTransactionMetaStoreHandler(transactionCoordinatorId, this);
	connectFuture.complete(null);
	return true;
	} else {
	State state = getState();
	cnx.channel().close();
	connectFuture.completeExceptionally(
	new IllegalStateException("Failed to change the state from " + state + " to Ready"));

[github-actions]

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