[PIP-379] DrainingHashEntry.refCount will increase when hashes move between existing consumers

[lhotari]

Background

• drainingHashesTracker.addEntry(c, stickyKeyHash) will increase the variable DrainingHashEntry.refCount.

Issue: the variable named DrainingHashEntry.refCount will be increased repeatedly at the following scenario:

• Register C1, assign k1 to C1
• Register C2, move k1 to C2
  • DrainingHashEntry.refCount increases
• Register C3, move k1 to C3
  • DrainingHashEntry.refCount increases

Originally posted by @poorbarcode in #23352 (comment)

[lhotari]

• Register C1, assign k1 to C1

• Register C2, move k1 to C2

  • DrainingHashEntry.refCount increases

• Register C3, move k1 to C3

  • DrainingHashEntry.refCount increases

This is already covered. While the hash is assigned to another consumer, the "draining hash" entry can only be assigned to a single consumer at a time. The reference count won't get increased for other consumers.

several locations:

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

Lines 197 to 201 in 38c1b79

	} else if (entry.getConsumer() != consumer) {
	throw new IllegalStateException(
	"Consumer " + entry.getConsumer() + " is already draining hash " + stickyHash
	+ " in dispatcher " + dispatcherName + ". Same hash being used for consumer " + consumer
	+ ".");

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

Lines 344 to 352 in d5795d2

	DrainingHashesTracker.DrainingHashEntry drainingHashEntry = drainingHashesTracker.getEntry(stickyKeyHash);
	if (drainingHashEntry != null && drainingHashEntry.getConsumer() != consumer) {
	log.warn("[{}] Another consumer id {} is already draining hash {}. Skipping adding {}:{} to pending acks "
	+ "for consumer {}. Adding the message to replay.",
	getName(), drainingHashEntry.getConsumer(), stickyKeyHash, ledgerId, entryId, consumer);
	addMessageToReplay(ledgerId, entryId, stickyKeyHash);
	// block message from sending
	return false;
	}

This is for the reassignment case:

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

Lines 294 to 302 in 38c1b79

	// hash has been reassigned to the original consumer, remove the entry
	// and don't block the hash
	if (entry.getConsumer() == consumer) {
	log.info("[{}] Hash {} has been reassigned consumer {}. "
	+ "The draining hash entry with refCount={} will be removed.",
	dispatcherName, stickyKeyHash, entry.getConsumer(), entry.refCount);
	drainingHashes.remove(stickyKeyHash, entry);
	return false;
	}

[lhotari]

Closing this as a non-issue. /cc @equanz