Broker registered the same consumers

[poorbarcode]

Describe the bug

If the client tries to subscribe multiple times over a short period of time, it is possible to have more than one same consumer at the same dispatcher( see #14970 ). just like this:

for ( long requestId = 1; i < 5; i++ ){
  ByteBuf request1 = Commands.newSubscribe(topic, subscription, consumerId, requestId , getSubType(),
          priorityLevel, consumerName, isDurable, startMessageIdData, metadata, readCompacted,
          conf.isReplicateSubscriptionState(),
          InitialPosition.valueOf(subscriptionInitialPosition.getValue()),
          startMessageRollbackDuration, si, createTopicIfDoesNotExist, conf.getKeySharedPolicy(),
          // Use the current epoch to subscribe.
          conf.getSubscriptionProperties(), CONSUMER_EPOCH.get(this));
  cnx.sendRequestWithId(request1, requestId).thenRun(() -> {});
}

Because there's something wrong with the ServerCnx.handleSubscribe:

if (existingConsumerFuture != null) {
    if (existingConsumerFuture.isDone() && !existingConsumerFuture.isCompletedExceptionally()) {
        Consumer consumer = existingConsumerFuture.getNow(null);
        commandSender.sendSuccessResponse(requestId);
        return null;
    } else {
        ServerError error = null;
        if (!existingConsumerFuture.isDone()) {
            error = ServerError.ServiceNotReady;
        } else {
            error = getErrorCode(existingConsumerFuture);
            // At this line, maybe delete a running future 
            consumers.remove(consumerId, existingConsumerFuture);
        }
        commandSender.sendErrorResponse(requestId, error, "Consumer is already present on the connection");
        return null;
    }
}

After problem , execute GET /admin/v2/persistent/public/default/my-topic/stats, response:

{
	"offloadedStorageSize": 0,
	"subscriptions": {
		"my-subscription": {
			"lastConsumedFlowTimestamp": 1649306204258,
			"nonContiguousDeletedMessagesRangesSerializedSize": 0,
			"msgRateExpired": 0.0,
			"lastMarkDeleteAdvancedTimestamp": 0,
			"msgOutCounter": 0,
			"backlogSize": 0,
			"type": "Shared",
			"durable": true,
			"bytesOutCounter": 0,
			"totalMsgExpired": 0,
			"nonContiguousDeletedMessagesRanges": 0,
			"consumersAfterMarkDeletePosition": {},
			"consumers": [{
					"metadata": {},
					"address": "\/192.168.0.115:62048",
					"msgRateRedeliver": 0.0,
					"avgMessagesPerEntry": 0,
					"msgOutCounter": 0,
					"clientVersion": "2.10.0-SNAPSHOT",
					"msgThroughputOut": 0.0,
					"availablePermits": 0,
					"bytesOutCounter": 0,
					"connectedSince": "2022-04-07T12:36:43.6431123+08:00",
					"blockedConsumerOnUnackedMsgs": false,
					"lastAckedTimestamp": 0,
					"unackedMessages": 0,
					"chunkedMessageRate": 0.0,
					"msgRateOut": 0.0,
					"consumerName": "a900f",
					"lastConsumedTimestamp": 0
				},
				{
					"metadata": {},
					"address": "\/192.168.0.115:62048",
					"msgRateRedeliver": 0.0,
					"avgMessagesPerEntry": 0,
					"msgOutCounter": 0,
					"clientVersion": "2.10.0-SNAPSHOT",
					"msgThroughputOut": 0.0,
					"availablePermits": 1000,
					"bytesOutCounter": 0,
					"connectedSince": "2022-04-07T12:36:44.2540793+08:00",
					"blockedConsumerOnUnackedMsgs": false,
					"lastAckedTimestamp": 0,
					"unackedMessages": 0,
					"chunkedMessageRate": 0.0,
					"msgRateOut": 0.0,
					"consumerName": "a900f",
					"lastConsumedTimestamp": 0
				}
			],
			"blockedSubscriptionOnUnackedMsgs": false,
			"msgDelayed": 0,
			"unackedMessages": 0,
			"chunkedMessageRate": 0,
			"msgBacklog": 0,
			"lastConsumedTimestamp": 0,
			"lastExpireTimestamp": 0,
			"msgRateRedeliver": 0.0,
			"isDurable": true,
			"msgThroughputOut": 0.0,
			"earliestMsgPublishTimeInBacklog": 0,
			"subscriptionProperties": {},
			"isReplicated": false,
			"allowOutOfOrderDelivery": false,
			"lastAckedTimestamp": 0,
			"msgBacklogNoDelayed": 0,
			"replicated": false,
			"msgRateOut": 0.0
		}
	},
	"nonContiguousDeletedMessagesRangesSerializedSize": 0,
	"lastOffloadFailureTimeStamp": 0,
	"msgOutCounter": 0,
	"backlogSize": 0,
	"bytesInCounter": 0,
	"lastOffloadLedgerId": 0,
	"bytesOutCounter": 0,
	"msgInCounter": 0,
	"nonContiguousDeletedMessagesRanges": 0,
	"publishers": [],
	"compaction": {
		"lastCompactionDurationTimeInMills": 0,
		"lastCompactionRemovedEventCount": 0,
		"lastCompactionSucceedTimestamp": 0,
		"lastCompactionFailedTimestamp": 0
	},
	"averageMsgSize": 0.0,
	"msgRateIn": 0.0,
	"replication": {},
	"earliestMsgPublishTimeInBacklogs": 0,
	"msgChunkPublished": false,
	"msgThroughputIn": 0.0,
	"msgThroughputOut": 0.0,
	"publishRateLimitedTimes": 0,
	"waitingPublishers": 0,
	"storageSize": 0,
	"deduplicationStatus": "Disabled",
	"lastOffloadSuccessTimeStamp": 0,
	"msgRateOut": 0.0
}

Close the consumer. retry execute GET /admin/v2/persistent/public/default/my-topic/stats, response:

{
	"offloadedStorageSize": 0,
	"subscriptions": {
		"my-subscription": {
			"lastConsumedFlowTimestamp": 1649306204258,
			"nonContiguousDeletedMessagesRangesSerializedSize": 0,
			"msgRateExpired": 0.0,
			"lastMarkDeleteAdvancedTimestamp": 0,
			"msgOutCounter": 0,
			"backlogSize": 0,
			"type": "Shared",
			"durable": true,
			"bytesOutCounter": 0,
			"totalMsgExpired": 0,
			"nonContiguousDeletedMessagesRanges": 0,
			"consumersAfterMarkDeletePosition": {},
			"consumers": [{
				"metadata": {},
				"address": "\/192.168.0.115:62048",
				"msgRateRedeliver": 0.0,
				"avgMessagesPerEntry": 0,
				"msgOutCounter": 0,
				"clientVersion": "2.10.0-SNAPSHOT",
				"msgThroughputOut": 0.0,
				"availablePermits": 1000,
				"bytesOutCounter": 0,
				"connectedSince": "2022-04-07T12:36:44.2540793+08:00",
				"blockedConsumerOnUnackedMsgs": false,
				"lastAckedTimestamp": 0,
				"unackedMessages": 0,
				"chunkedMessageRate": 0.0,
				"msgRateOut": 0.0,
				"consumerName": "a900f",
				"lastConsumedTimestamp": 0
			}],
			"blockedSubscriptionOnUnackedMsgs": false,
			"msgDelayed": 0,
			"unackedMessages": 0,
			"chunkedMessageRate": 0,
			"msgBacklog": 0,
			"lastConsumedTimestamp": 0,
			"lastExpireTimestamp": 0,
			"msgRateRedeliver": 0.0,
			"isDurable": true,
			"msgThroughputOut": 0.0,
			"earliestMsgPublishTimeInBacklog": 0,
			"subscriptionProperties": {},
			"isReplicated": false,
			"allowOutOfOrderDelivery": false,
			"lastAckedTimestamp": 0,
			"msgBacklogNoDelayed": 0,
			"replicated": false,
			"msgRateOut": 0.0
		}
	},
	"nonContiguousDeletedMessagesRangesSerializedSize": 0,
	"lastOffloadFailureTimeStamp": 0,
	"msgOutCounter": 0,
	"backlogSize": 0,
	"bytesInCounter": 0,
	"lastOffloadLedgerId": 0,
	"bytesOutCounter": 0,
	"msgInCounter": 0,
	"nonContiguousDeletedMessagesRanges": 0,
	"publishers": [],
	"compaction": {
		"lastCompactionDurationTimeInMills": 0,
		"lastCompactionRemovedEventCount": 0,
		"lastCompactionSucceedTimestamp": 0,
		"lastCompactionFailedTimestamp": 0
	},
	"averageMsgSize": 0.0,
	"msgRateIn": 0.0,
	"replication": {},
	"earliestMsgPublishTimeInBacklogs": 0,
	"msgChunkPublished": false,
	"msgThroughputIn": 0.0,
	"msgThroughputOut": 0.0,
	"publishRateLimitedTimes": 0,
	"waitingPublishers": 0,
	"storageSize": 0,
	"deduplicationStatus": "Disabled",
	"lastOffloadSuccessTimeStamp": 0,
	"msgRateOut": 0.0
}

execute DELETE /admin/v2/persistent/public/default/my-topic/subscription/my-subscription. response timeout

execute bin/pulsar-admin topics unsubscribe --force --subscription my-subscription my-topic will get 'Reason: java.util.concurrent.TimeoutException'

ralate #13787

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[poorbarcode]

fixed by #15051