[Bug] Dispatcher will be closing and reopening when Key_shared consumers have different policies (AUTO_SPLIT/STICKY, allowOutOfOrderDelivery true/false)

[lhotari]

Search before asking

• 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.

Version

all released versions including master branch

Minimal reproduce step

The conclusion is currently based on the source code:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

Lines 287 to 309 in 766d2a4

	case Key_Shared:
	KeySharedMeta ksm = consumer.getKeySharedMeta();
	if (dispatcher == null || dispatcher.getType() != SubType.Key_Shared
	|| !((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher)
	.hasSameKeySharedPolicy(ksm)) {
	previousDispatcher = dispatcher;
	dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(topic, cursor, this,
	topic.getBrokerService().getPulsar().getConfiguration(), ksm);
	}
	break;
	default:
	return FutureUtil.failedFuture(
	new ServerMetadataException("Unsupported subscription type"));
	}
	if (previousDispatcher != null) {
	previousDispatcher.close().thenRun(() -> {
	log.info("[{}][{}] Successfully closed previous dispatcher", topicName, subName);
	}).exceptionally(ex -> {
	log.error("[{}][{}] Failed to close previous dispatcher", topicName, subName, ex);
	return null;
	});
	}

What did you expect to see?

When multiple consumers are using different policies, this should be properly handled.
One possibility is to keep the policy of the connected consumers and reject any other consumers and return a proper error message.

What did you see instead?

Based on the source code, it looks like the solution cannot work:

• a new dispatcher is created while the existing dispatcher is active
• the existing dispatcher is closed asynchronously
• no proper errors messages are logged or logged that could help resolving the situation

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[pdolif]

@lhotari I tried to implement this: pdolif@e241cff
When the policy of a new consumer is incompatible with the existing consumers of the subscription it is rejected.

I also wrote a test but it works only when disabling the transaction coordinator in the setup() method. I could not find out how to make the test work with the transaction coordinator being enabled. When calling addConsumer() on the subscription the pendingAckHandleFuture seems to block.

[lhotari]

@lhotari I tried to implement this: pdolif@e241cff When the policy of a new consumer is incompatible with the existing consumers of the subscription it is rejected.

I also wrote a test but it works only when disabling the transaction coordinator in the setup() method. I could not find out how to make the test work with the transaction coordinator being enabled. When calling addConsumer() on the subscription the pendingAckHandleFuture seems to block.

makes sense. good work, @pdolif! Thanks for the contribution.