[fix][transaction] Fix potentially unfinishable future.

[mattisonchao]

Motivation

The relative code is as below:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java

Lines 167 to 250 in 9861dfb

	public CompletableFuture<Void> handleTcClientConnect(TransactionCoordinatorID tcId) {
	CompletableFuture<Void> completableFuture = new CompletableFuture<>();
	internalPinnedExecutor.execute(() -> {
	if (stores.get(tcId) != null) {
	completableFuture.complete(null);
	} else {
	pulsarService.getBrokerService().checkTopicNsOwnership(TopicName
	.TRANSACTION_COORDINATOR_ASSIGN.getPartition((int) tcId.getId()).toString())
	.thenRun(() -> internalPinnedExecutor.execute(() -> {
	final Semaphore tcLoadSemaphore = this.tcLoadSemaphores
	.computeIfAbsent(tcId.getId(), (id) -> new Semaphore(1));
	Deque<CompletableFuture<Void>> deque = pendingConnectRequests
	.computeIfAbsent(tcId.getId(), (id) -> new ConcurrentLinkedDeque<>());
	if (tcLoadSemaphore.tryAcquire()) {
	// when tcLoadSemaphore.release(), this command will acquire semaphore,
	// so we should jude the store exist again.
	if (stores.get(tcId) != null) {
	completableFuture.complete(null);
	tcLoadSemaphore.release();
	return;
	}
	openTransactionMetadataStore(tcId).thenAccept((store) -> internalPinnedExecutor.execute(() -> {
	stores.put(tcId, store);
	LOG.info("Added new transaction meta store {}", tcId);
	long endTime = System.currentTimeMillis() + HANDLE_PENDING_CONNECT_TIME_OUT;
	while (true) {
	// prevent thread in a busy loop.
	if (System.currentTimeMillis() < endTime) {
	CompletableFuture<Void> future = deque.poll();
	if (future != null) {
	// complete queue request future
	future.complete(null);
	} else {
	break;
	}
	} else {
	deque.clear();
	break;
	}
	}
	completableFuture.complete(null);
	tcLoadSemaphore.release();
	})).exceptionally(e -> {
	internalPinnedExecutor.execute(() -> {
	completableFuture.completeExceptionally(e.getCause());
	// release before handle request queue,
	//in order to client reconnect infinite loop
	tcLoadSemaphore.release();
	long endTime = System.currentTimeMillis() + HANDLE_PENDING_CONNECT_TIME_OUT;
	while (true) {
	// prevent thread in a busy loop.
	if (System.currentTimeMillis() < endTime) {
	CompletableFuture<Void> future = deque.poll();
	if (future != null) {
	// this means that this tc client connection connect fail
	future.completeExceptionally(e);
	} else {
	break;
	}
	} else {
	deque.clear();
	break;
	}
	}
	LOG.error("Add transaction metadata store with id {} error", tcId.getId(), e);
	});
	return null;
	});
	} else {
	// only one command can open transaction metadata store,
	// other will be added to the deque, when the op of openTransactionMetadataStore finished
	// then handle the requests witch in the queue
	deque.add(completableFuture);
	if (LOG.isDebugEnabled()) {
	LOG.debug("Handle tc client connect added into pending queue! tcId : {}", tcId.toString());
	}
	}
	}));
	}
	});
	return completableFuture;
	}

When checkTopicNsOwnership throws an exception, it will cause the completableFuture never complete.

This PR is just to fix this unfinishable future issue, after that, I will try to refactor this part of the code to make it not so complex.

Modifications

• Add checkTopicNsOwnership exception catch.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• no-need-doc