[fix][broker] Add `internalAddConsumer` to help track StackOverflowError issue by Technoboy- · Pull Request #24920 · apache/pulsar

Technoboy- · 2025-10-30T05:11:29Z

Motivation

We see the below StackOverflowError log, but we don't have the heapdump for it, so try to add internalAddConsumer to log to help find the root cause.

2025-10-28T22:54:08,431+0000 [configuration-metadata-store-14-1] ERROR org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/test-partition-0] Failed to create subscription: test-subscription
java.util.concurrent.CompletionException: java.lang.StackOverflowError
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1194) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171) 
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171)
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171) 
	at 
...
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171) 
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171)
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171)
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176)
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.addConsumer(AbstractDispatcherSingleActiveConsumer.java:171) 
	at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.lambda$addConsumer$2(AbstractDispatcherSingleActiveConsumer.java:176) 
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2309) ~[?:?]

Documentation

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

lhotari · 2025-10-30T06:01:56Z

+                        try {
+                            removeConsumer(actConsumer);
+                        } catch (BrokerServiceException e) {
+                            log.warn("[{}] Remove inactive exclusive consumer {}", this.topicName, consumer);
+                        }


I don't think that this is the correct solution. The consumer gets removed when the connection is closed:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Lines 438 to 462 in acad78c

consumers.forEach((__, consumerFuture) -> {

// prevent race conditions in completing consumers

if (!consumerFuture.isDone()

&& consumerFuture.completeExceptionally(new IllegalStateException("Connection closed."))) {

return;

}

if (consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {

Consumer consumer = consumerFuture.getNow(null);

try {

consumer.close();

if (brokerInterceptor != null) {

brokerInterceptor.consumerClosed(this, consumer, consumer.getMetadata());

}

} catch (BrokerServiceException e) {

log.warn("Consumer {} was already closed: {}", consumer, e);

}

}

});

this.topicListService.inactivate();

this.service.getPulsarStats().recordConnectionClose();

// complete possible pending connection check future

if (connectionCheckInProgress != null && !connectionCheckInProgress.isDone()) {

connectionCheckInProgress.complete(Optional.of(false));

}

Closing gets triggered in the liveness check:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Lines 3715 to 3721 in acad78c

if (finalConnectionCheckInProgress == connectionCheckInProgress) {

/**

* {@link #connectionCheckInProgress} will be completed when

* {@link #channelInactive(ChannelHandlerContext)} event occurs, so skip set it here.

*/

log.warn("[{}] Connection check timed out. Closing connection.", this.toString());

ctx.close();

I'd suppose that there's a race condition that isn't properly handled.

checkConnectionLiveness returned false.

The active consumer was not removed.

Seems we should fix the behaviour of checkConnectionLiveness or remove consumers if the connection is inactive

yes, but no dump file now. I just raise this patch to test if there are some tests can't pass

@lhotari @poorbarcode Could you please check if there's any other information that needs to be added?

@lhotari I think the previous change here is correct: bcb763b

channelInactive is not synchronized with addConsumer, so there is a race condition.

isActive = false is called:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Line 411 in 402ed5b

isActive = false;

addConsumer is called on a new consumer

consumer.close() is called:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Line 447 in 402ed5b

consumer.close();

Then step 2 will always fail with StackOverflowError.

2025-10-28T22:54:08,431+0000 [configuration-metadata-store-14-1] ERROR org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/test-partition-0] Failed to create subscription: test-subscription java.util.concurrent.CompletionException: java.lang.StackOverflowError

From the thread name, we can see the StackOverflowError error happens in the metadata store thread. However, if the consumer's connection (ServerCnx) is active, the future will always complete in a Netty event loop thread, whose name starts with pulsar-io.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Line 3697 in 402ed5b

return NettyFutureUtil.toCompletableFuture(ctx.executor().submit(() -> {

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

Line 3707 in 402ed5b

ctx.executor().schedule(() -> {

After thinking for a while, removing the consumer immediately might not be a good idea, I opened a PR that adds a test and a retry based solution. PTAL @poorbarcode @Technoboy- @lhotari #24934

codecov-commenter · 2025-10-30T06:22:59Z

Codecov Report

❌ Patch coverage is 44.44444% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.22%. Comparing base (f55d45a) to head (3db463d).
⚠️ Report is 7 commits behind head on master.

Files with missing lines	Patch %	Lines
...ervice/AbstractDispatcherSingleActiveConsumer.java	44.44%	4 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #24920       +/-   ##
=============================================
+ Coverage     38.74%   74.22%   +35.48%     
- Complexity    13265    33866    +20601     
=============================================
  Files          1856     1913       +57     
  Lines        145165   149433     +4268     
  Branches      16848    17359      +511     
=============================================
+ Hits          56238   110923    +54685     
+ Misses        81385    29642    -51743     
- Partials       7542     8868     +1326

Flag	Coverage Δ
inttests	`26.43% <11.11%> (-0.19%)`	⬇️
systests	`22.67% <11.11%> (-0.22%)`	⬇️
unittests	`73.76% <44.44%> (+38.88%)`	⬆️

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

Files with missing lines	Coverage Δ
...ervice/AbstractDispatcherSingleActiveConsumer.java	`87.21% <44.44%> (+6.41%)`	⬆️

... and 1411 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

lhotari · 2025-11-03T11:11:29Z

+            log.warn("Added consumer failed with attempt count {}, consumers: {}, active consumer: {},"
+                            + "active state : {}",
+                    addConsumerFailedAttemptCount, consumers, actConsumer, actConsumer.cnx().isActive());
+            addConsumerFailedAttemptCount = 0;


Add code to reset the counter also when the consumer has been successfully added

Perhaps a solution where the counter is passed as a parameter in the call stack would be more reliable?

Duplicated

lhotari · 2025-11-03T15:46:20Z

Problem fixed by #24934

Remove inactive exclusive consumer after checking liveness

bcb763b

Technoboy- added the ready-to-test label Oct 30, 2025

Technoboy- added this to the 4.2.0 milestone Oct 30, 2025

Technoboy- self-assigned this Oct 30, 2025

Technoboy- added release/3.0.9 release/3.3.9 release/4.1.2 release/4.0.8 labels Oct 30, 2025

github-actions Bot added the doc-not-needed Your PR changes do not impact docs label Oct 30, 2025

lhotari reviewed Oct 30, 2025

View reviewed changes

update

00921e0

poorbarcode approved these changes Nov 3, 2025

View reviewed changes

Technoboy- changed the title ~~[fix][broker] Remove inactive exclusive consumer after checking liveness~~ [fix][broker] Add internalAddConsumer to help track StackOverflowError issue Nov 3, 2025

BewareMyPower reviewed Nov 3, 2025

View reviewed changes

Comment thread ...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java Outdated

Comment thread ...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java Outdated

lhotari reviewed Nov 3, 2025

View reviewed changes

Comment thread ...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java Outdated

address comment

3db463d

lhotari reviewed Nov 3, 2025

View reviewed changes

This comment was marked as duplicate.

Sign in to view

BewareMyPower mentioned this pull request Nov 3, 2025

[fix][broker] Fix stack overflow caused by race condition when closing a connection #24934

Merged

4 tasks

lhotari closed this Nov 3, 2025

lhotari added release/3.3.10 release/3.0.16 and removed release/3.0.9 release/3.3.9 labels Nov 11, 2025

	consumers.forEach((__, consumerFuture) -> {
	// prevent race conditions in completing consumers
	if (!consumerFuture.isDone()
	&& consumerFuture.completeExceptionally(new IllegalStateException("Connection closed."))) {
	return;
	}
	if (consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
	Consumer consumer = consumerFuture.getNow(null);
	try {
	consumer.close();
	if (brokerInterceptor != null) {
	brokerInterceptor.consumerClosed(this, consumer, consumer.getMetadata());
	}
	} catch (BrokerServiceException e) {
	log.warn("Consumer {} was already closed: {}", consumer, e);
	}
	}
	});
	this.topicListService.inactivate();
	this.service.getPulsarStats().recordConnectionClose();

	// complete possible pending connection check future
	if (connectionCheckInProgress != null && !connectionCheckInProgress.isDone()) {
	connectionCheckInProgress.complete(Optional.of(false));
	}

	if (finalConnectionCheckInProgress == connectionCheckInProgress) {
	/**
	* {@link #connectionCheckInProgress} will be completed when
	* {@link #channelInactive(ChannelHandlerContext)} event occurs, so skip set it here.
	*/
	log.warn("[{}] Connection check timed out. Closing connection.", this.toString());
	ctx.close();

Uh oh!

Conversation

Technoboy- commented Oct 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Documentation

Uh oh!

lhotari Oct 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

Technoboy- Oct 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Technoboy- Nov 3, 2025

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Nov 3, 2025

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Nov 3, 2025

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Oct 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lhotari Nov 3, 2025

Choose a reason for hiding this comment

Uh oh!

lhotari Nov 3, 2025

Choose a reason for hiding this comment

Uh oh!

This comment was marked as duplicate.

Uh oh!

lhotari commented Nov 3, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Technoboy- commented Oct 30, 2025 •

edited

Loading

lhotari Oct 30, 2025 •

edited

Loading

Technoboy- Oct 30, 2025 •

edited

Loading

codecov-commenter commented Oct 30, 2025 •

edited

Loading