[Bug] Partial partition consumers silent after topic fencing

[nodece]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

• Observed version: Pulsar 2.10.1 (unsupported)
• Likely affected: All Pulsar versions, including 3.x, 4.x

Issue Description

When a topic is fenced, some partition consumers may silently stop receiving messages, while others continue working normally.

It only occurs under specific timing and failure conditions, which makes it hard to detect and debug:

• The broker sometimes fails to deliver the close command to certain consumers (e.g. due to transient network issues, broker overload, or connection state races).
• Consumers rely entirely on broker-initiated notifications and do not actively verify fencing state after subscription is established.

Once this happens, the affected partition consumer enters a permanent silent failure state.

Symptoms

• ✅ TCP connection remains alive
• ✅ No exceptions or error logs
• ❌ Some partitions stop receiving messages indefinitely

Meanwhile, other partitions of the same consumer continue to work normally, which further hides the issue.

Error messages

Reproducing the issue

1. Create a partitioned topic and subscribe a consumer.
2. Trigger topic fencing via topic unload.
3. Publish messages to the topic.
4. Observe that some partitions stop receiving messages while others continue normally, with no errors or logs.

This issue is not 100% reproducible.

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

• The broker sometimes fails to deliver the fencing command to certain consumers (e.g. due to transient network issues, broker overload, or connection state races).

@nodece What does "deliver the fencing command" mean in practice? Is there a "fencing command" in Pulsar?

[lhotari]

Perhaps #21082 is related? IIRC, there have been also other similar issues that have been reported in the past.

[lhotari]

A long time ago I worked on #20540 which was never merged. There have been many changes after that so most likely the PR is not up-to-date anymore.
At that time, I was able to reproduce issues by unloading a topic multiple times. I used https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioUnloading.java for doing that.

[nodece]

@lhotari Thanks for pointing out the related issues! The fact that #21082 and similar issues have been reported multiple times, along with the previous attempt in PR #20540, confirms this is a systemic problem rather than an isolated edge case.

Root cause

The fundamental issue is the reliance on one-way broker-to-consumer notifications. During topic unloading (as demonstrated in the test scenario), there's a critical timing window where:

• Broker-side state transitions rapidly: fencing → unloading → close consumer → cleanup cache
• The close notification can be lost, consumer connections may not receive the notification before being closed, leaving them in a "zombie" state

Proposed Solution

Instead of fixing the complex broker notification timing issues, a more robust approach would be:

Add a heartbeat mechanism at the consumer level.

This would allow consumers to:

• Proactively detect subscription issues
• Auto-recover from "zombie" states without waiting for broker notifications
• Handle topic transfers and unloading gracefully

Trade-offs

Disadvantage: This introduces additional overhead (network traffic, CPU cycles on both broker and consumer sides).

Mitigation:

• Make heartbeat interval configurable (e.g., 30s-60s by default)
• Only enable for critical consumers or make it opt-in
• Piggyback heartbeat with existing traffic when possible

This aligns with distributed systems best practices where active health checking is more reliable than passive notification, though it comes at a performance cost.

[lhotari]

• The close notification can be lost, consumer connections may not receive the notification before being closed, leaving them in a "zombie" state

How could this happen? In TCP/IP, the connection does handle retries at the TCP/IP stack level. The connection might die, but that's why there is the Pulsar protocol level heartbeat with ping-pong messages. The Pulsar client should remove the consumer when the connection is closed and then retry.

[nodece]

How could this happen?

Under high broker load or in extreme timing conditions, the broker might fail to deliver the close consumer command to the client, even though the TCP connection itself remains healthy. If the connection is disconnected, the client would detect it via the Pulsar heartbeat and automatically resubscribe in the background.

This issue impacts our customers in production. While updating the broker TLS version might improve reliability slightly, it cannot fully prevent this scenario. Just sharing my observation.

[lhotari]

Under high broker load or in extreme timing conditions, the broker might fail to deliver the close consumer command to the client, even though the TCP connection itself remains healthy.

I don't immediately see how it could be possible that the delivery of the command would fail when the TCP connection itself remains healthy.

It's possible that the delivery is delayed, that's a valid case. However, if the delivery fails, that would mean that there's another bug that should be addressed directly. In the org.apache.pulsar.broker.service.Consumer class, there are 2 different methods related to closing: close and disconnect. When close is called, it won't notify the client. Perhaps somewhere the call to disconnect is missing and close is used instead?

The close notifications are sent at

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

Lines 3388 to 3400 in b1019ce

	private void closeConsumer(long consumerId, Optional<BrokerLookupData> assignedBrokerLookupData) {
	if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {
	assignedBrokerLookupData.ifPresentOrElse(lookup -> {
	LookupData lookupData = getLookupData(lookup);
	writeAndFlush(Commands.newCloseConsumer(consumerId, -1L,
	lookupData.getBrokerUrl(),
	lookupData.getBrokerUrlTls()));
	},
	() -> writeAndFlush(Commands.newCloseConsumer(consumerId, -1L, null, null)));
	} else {
	close();
	}
	}

It would be possible to use ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) to close the connection immediately if writing to the channel fails. The reason why .addListener isn't used in all writes is to avoid creating a ChannelPromise instance for all invocations to writeAndFlush. Instead, ctx.voidPromise() is used in org.apache.pulsar.common.util.netty.NettyChannelUtil#writeAndFlushWithVoidPromise.

[nodece]

Sounds good, I will check this next week