[fix][client] Fix non partitioned topic consumer call RedeliverUnacknowledgedMessages cause deadlock.

[Technoboy-]

Motivation

#17318 has made internalRedeliverUnacknowledgedMessages executed by internalPinnedExecutor.

If the consumer received msg then call redeliverUnacknowledgedMessages like the below :

consumer.receiveAsync()      // using internalPinnedExecutor thread
                .thenCompose(m -> {
                    return consumer.acknowledgeCumulativeAsync(m);
                })
                .thenAccept(ignore -> {
                    try {
                        consumer.redeliverUnacknowledgedMessages(); // still the same internalPinnedExecutor thread, it will be blocked
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                })

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

Lines 1915 to 1921 in dcf0b60

	public void redeliverUnacknowledgedMessages() {
	try {
	internalRedeliverUnacknowledgedMessages().get();
	} catch (ExecutionException e) {
	throw e.getCause();
	}
	}

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

Lines 1861 to 1911 in dcf0b60

	public CompletableFuture<Void> internalRedeliverUnacknowledgedMessages() {
	return CompletableFuture.runAsync(() -> {
	// First : synchronized in order to handle consumer reconnect produce race condition, when broker receive
	// redeliverUnacknowledgedMessages and consumer have not be created and then receive reconnect epoch
	// change the broker is smaller than the client epoch, this will cause client epoch smaller
	// than broker epoch forever. client will not receive message anymore.
	// Second : we should synchronized `ClientCnx cnx = cnx()` to prevent use old cnx to
	// send redeliverUnacknowledgedMessages to a old broker
	synchronized (ConsumerImpl.this) {
	ClientCnx cnx = cnx();
	// V1 don't support redeliverUnacknowledgedMessages
	if (cnx != null && cnx.getRemoteEndpointProtocolVersion() < ProtocolVersion.v2.getValue()) {
	if ((getState() == State.Connecting)) {
	log.warn("[{}] Client Connection needs to be established "
	+ "for redelivery of unacknowledged messages", this);
	} else {
	log.warn("[{}] Reconnecting the client to redeliver the messages.", this);
	cnx.ctx().close();
	}
	return;
	}
	// we should increase epoch every time, because MultiTopicsConsumerImpl also increase it,
	// we need to keep both epochs the same
	if (conf.getSubscriptionType() == SubscriptionType.Failover
	|| conf.getSubscriptionType() == SubscriptionType.Exclusive) {
	CONSUMER_EPOCH.incrementAndGet(this);
	}
	// clear local message
	int currentSize = incomingMessages.size();
	clearIncomingMessages();
	unAckedMessageTracker.clear();
	// is channel is connected, we should send redeliver command to broker
	if (cnx != null && isConnected(cnx)) {
	cnx.ctx().writeAndFlush(Commands.newRedeliverUnacknowledgedMessages(
	consumerId, CONSUMER_EPOCH.get(this)), cnx.ctx().voidPromise());
	if (currentSize > 0) {
	increaseAvailablePermits(cnx, currentSize);
	}
	if (log.isDebugEnabled()) {
	log.debug("[{}] [{}] [{}] Redeliver unacked messages and send {} permits", subscription, topic,
	consumerName, currentSize);
	}
	} else {
	log.warn("[{}] Send redeliver messages command but the client is reconnect or close, "
	+ "so don't need to send redeliver command to broker", this);
	}
	}
	}, internalPinnedExecutor);
	}

Modification

• Remove using internalPinnedExecutor to execute internalRedeliverUnacknowledgedMessages.

Documentation

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

[codelipenghui]

@Technoboy- I think the right fix should change

public void redeliverUnacknowledgedMessages() { 
     try { 
         internalRedeliverUnacknowledgedMessages().get(); 
     } catch (ExecutionException e) { 
         throw e.getCause(); 
     } 
 } 

to

public void redeliverUnacknowledgedMessages() { 
         internalRedeliverUnacknowledgedMessages();
 } 

I'm afraid we will introduce a potential race condition issue if we remove the internalPinnedExecutor for internalRedeliverUnacknowledgedMessages

[codelipenghui]

Oh sorry. But it will introduce another problem. Users have completed the redeliverUnacknowledgedMessages() and then try to call receive() again. Due to redeliverUnacknowledgedMessages() being executed in another thread. Users can get the message before cleaning up the incoming queue.

[codecov-commenter]

Codecov Report

Merging #18462 (43966a1) into master (fdf86d3) will decrease coverage by 9.48%.
The diff coverage is 5.00%.

@@             Coverage Diff              @@
##             master   #18462      +/-   ##
============================================
- Coverage     45.67%   36.18%   -9.49%     
+ Complexity    10075     7734    -2341     
============================================
  Files           693      697       +4     
  Lines         67940    67979      +39     
  Branches       7273     7278       +5     
============================================
- Hits          31030    24600    -6430     
- Misses        33333    40076    +6743     
+ Partials       3577     3303     -274     

Flag	Coverage Δ
unittests	36.18% <5.00%> (-9.49%)	⬇️

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

Impacted Files	Coverage Δ
...adbalance/extensions/channel/ServiceUnitState.java	0.00% <0.00%> (ø)
...lance/extensions/channel/ServiceUnitStateData.java	0.00% <0.00%> (ø)
...ar/broker/loadbalance/extensions/models/Split.java	0.00% <0.00%> (ø)
...r/broker/loadbalance/extensions/models/Unload.java	0.00% <0.00%> (ø)
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	15.05% <0.00%> (-0.05%)	⬇️
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	22.99% <25.00%> (+0.03%)	⬆️
...a/org/apache/pulsar/client/impl/MessageIdImpl.java	80.55% <33.33%> (+18.05%)	⬆️
...apache/pulsar/client/impl/ConsumerBuilderImpl.java	27.80% <50.00%> (+0.29%)	⬆️
...ava/org/apache/pulsar/broker/admin/v1/Brokers.java	0.00% <0.00%> (-100.00%)	⬇️
...va/org/apache/pulsar/broker/admin/v1/Clusters.java	0.00% <0.00%> (-100.00%)	⬇️
... and 154 more

[codelipenghui]

The issue should also happen on a partitioned topic, right?

public void redeliverUnacknowledgedMessages() {
        List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
        internalPinnedExecutor.execute(() -> {
            CONSUMER_EPOCH.incrementAndGet(this);
            consumers.values().stream().forEach(consumer -> {
                futures.add(consumer.internalRedeliverUnacknowledgedMessages());
                consumer.unAckedChunkedMessageIdSequenceMap.clear();
            });
            clearIncomingMessages();
            unAckedMessageTracker.clear();
        });
        try {
            FutureUtil.waitForAll(futures).get();
        } catch (ExecutionException e) {
            throw e.getCause();
        }
        resumeReceivingFromPausedConsumersIfNeeded();
    }

[Technoboy-]

Close this patch due to the revert of #18475