[Broker] Fix race conditions in closing producers and consumers

[lhotari]

Motivation

• closing ServerCnx while producers or consumers are created can lead to a producer or consumer never getting removed from the topic's list of producers

Modifications

If the future isn't completed by the time of closing the ServerCnx, complete it exceptionally. This will ensure that the existing logic in creating a producer or consumer doesn't leave the just created producer or consumer open.

[michaeljmarshall]

LGTM, other than a couple nits.

It's worth pointing out that the race is very subtle and comes here:

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

Lines 1302 to 1328 in 636d5b4

	if (isActive()) {
	if (producerFuture.complete(producer)) {
	log.info("[{}] Created new producer: {}", remoteAddress, producer);
	commandSender.sendProducerSuccessResponse(requestId, producerName,
	producer.getLastSequenceId(), producer.getSchemaVersion(),
	newTopicEpoch, true /* producer is ready now */);
	if (getBrokerService().getInterceptor() != null) {
	getBrokerService().getInterceptor().
	producerCreated(this, producer, metadata);
	}
	return;
	} else {
	// The producer's future was completed before by
	// a close command
	producer.closeNow(true);
	log.info("[{}] Cleared producer created after"
	+ " timeout on client side {}",
	remoteAddress, producer);
	}
	} else {
	producer.closeNow(true);
	log.info("[{}] Cleared producer created after connection was closed: {}",
	remoteAddress, producer);
	producerFuture.completeExceptionally(
	new IllegalStateException(
	"Producer created after connection was closed"));
	}

[merlimat]

👍