[fix][broker] Prevent repeated consumer close on topic migration

[Denovo1998]

Motivation

PR #26075 fixed the non-persistent topic subscribe race by redirecting a migrated consumer before adding it to the subscription. However, that path still completes the subscribe future with the same consumer, and ServerCnx.handleSubscribe runs the generic post-subscribe migration check again.

That second check can redirect and disconnect the same consumer a second time. For non-persistent topics this can double-decrement the topic usage count and emit duplicate TopicMigrated commands.

Modifications

• Make Consumer.topicMigrated(...) idempotent for each consumer with an atomic once guard.
• Keep the first migration handling unchanged: send TopicMigrated and disconnect the consumer.
• Ignore repeated migration handling for the same consumer so the ServerCnx post-subscribe check remains safe after an inline non-persistent redirect.
• Add a regression test that simulates the post-subscribe migration check after NonPersistentTopic.internalSubscribe already redirected the consumer, verifying one redirect and balanced topic usage count.

Verifying this change

• Make sure that the change passes the CI checks.

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopicTest.testPostSubscribeMigrationCheckDoesNotDoubleCloseRedirectedConsumer

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopicTest.testSubscribeOnMigratedTopicSkipsAddingConsumer --tests org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopicTest.testPostSubscribeMigrationCheckDoesNotDoubleCloseRedirectedConsumer

• ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[lhotari]

It seems like the original logic before #26075 is incorrect. Consumer.topicMigrated should only get called by this code:

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

Lines 2017 to 2027 in 808133d

	.thenComposeAsync(consumer -> consumer.checkAndApplyTopicMigrationAsync()
	.thenAcceptAsync(migrated -> {
	if (migrated) {
	log.info()
	.attr("consumerId", consumerId)
	.attr("subscription", subscriptionName)
	.attr("topic", topicName)
	.log("Disconnecting consumer on migrated subscription on topic");
	consumers.remove(consumerId, consumerFuture);
	return;
	}

One possibility would be to add this method to Consumer so that it could be used in NonPersistentTopic.internalSubscribe instead of using checkAndApplyTopicMigrationAsync

    public CompletableFuture<Boolean> checkTopicMigrationAsync() {
        if (!subscription.isSubscriptionMigrated()) {
            return CompletableFuture.completedFuture(false);
        }
        return AbstractTopic.getMigratedClusterUrlAsync(cnx.getBrokerService().getPulsar(), topicName)
                .thenApply(clusterUrl -> {
                    return clusterUrl.isPresent();
                });
    }

[Denovo1998]

It seems like the original logic before #26075 is incorrect. Consumer.topicMigrated should only get called by this code:

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

Lines 2017 to 2027 in 808133d

	.thenComposeAsync(consumer -> consumer.checkAndApplyTopicMigrationAsync()
	.thenAcceptAsync(migrated -> {
	if (migrated) {
	log.info()
	.attr("consumerId", consumerId)
	.attr("subscription", subscriptionName)
	.attr("topic", topicName)
	.log("Disconnecting consumer on migrated subscription on topic");
	consumers.remove(consumerId, consumerFuture);
	return;
	}

One possibility would be to add this method to Consumer so that it could be used in NonPersistentTopic.internalSubscribe instead of using checkAndApplyTopicMigrationAsync

    public CompletableFuture<Boolean> checkTopicMigrationAsync() {
        if (!subscription.isSubscriptionMigrated()) {
            return CompletableFuture.completedFuture(false);
        }
        return AbstractTopic.getMigratedClusterUrlAsync(cnx.getBrokerService().getPulsar(), topicName)
                .thenApply(clusterUrl -> {
                    return clusterUrl.isPresent();
                });
    }

@lhotari
Thanks for the suggestion. I updated the PR to keep Consumer.topicMigrated(...) applied only from the common ServerCnx post-subscribe path.

NonPersistentTopic.internalSubscribe now uses a new pure Consumer.checkTopicMigrationAsync() helper to detect migrated subscriptions and skip addConsumerToSubscription without sending the redirect or closing the consumer. The existing checkAndApplyTopicMigrationAsync() still owns the actual redirect/close behavior.