[fix][client] In method LastCumulativeAck#update, bitSetRecyclable does not update when messageId is equal.

[lordcheng10]

Motivation

In method LastCumulativeAck#update, bitSetRecyclable does not update when messageId is equal:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

Lines 660 to 669 in fa328a4

	public synchronized void update(final MessageIdImpl messageId, final BitSetRecyclable bitSetRecyclable) {
	if (messageId.compareTo(this.messageId) > 0) {
	if (this.bitSetRecyclable != null && this.bitSetRecyclable != bitSetRecyclable) {
	this.bitSetRecyclable.recycle();
	}
	set(messageId, bitSetRecyclable);
	flushRequired = true;
	}
	}

Modifications

When the messageId is equal, but the batch index is larger, the bitSetRecyclable should be updated：

   public synchronized void update(final MessageIdImpl messageId, final BitSetRecyclable bitSetRecyclable) {
        if (messageId.equals(this.messageId)) {
            if (this.bitSetRecyclable != null && bitSetRecyclable != null
                    && bitSetRecyclable.nextSetBit(0) > this.bitSetRecyclable.nextSetBit(0)) {
                this.bitSetRecyclable.recycle();
                set(messageId, bitSetRecyclable);
                flushRequired = true;
            }
            return;
        }

        if (messageId.compareTo(this.messageId) > 0) {
            if (this.bitSetRecyclable != null && this.bitSetRecyclable != bitSetRecyclable) {
                this.bitSetRecyclable.recycle();
            }
            set(messageId, bitSetRecyclable);
            flushRequired = true;
        }
    }

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: lordcheng10#36

[BewareMyPower]

We already have the compare sult, I think we can improve it like

        int compareResult = messageId.compareTo(this.messageId);
        if (messageId.compareTo(this.messageId) > 0) {
            /* ... */
        } else if (compareResult == 0) {
            /* compare bitSetRecyclable... */
        }

[Jason918]

+1

[congbobo184]

seem compareTo has already compare the batchIndex.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdImpl.java

Lines 248 to 253 in fa328a4

	return ComparisonChain.start()
	.compare(ledgerId1, ledgerId2)
	.compare(entryId1, entryId2)
	.compare(partitionIndex1, partitionIndex2)
	.compare(batchIndex1, batchIndex2)
	.result();

so we don't need to do single check, right?

[BewareMyPower]

+1

[lordcheng10]

Yes, you are right

[BewareMyPower]

Actually I have another question. Could it happen in a real world case?

When the BitSetRecyclable object is not null, the messageId argument must be a BatchMessageIdImpl.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

Lines 378 to 381 in fa328a4

	BitSetRecyclable bitSet = BitSetRecyclable.create();
	bitSet.set(0, batchMessageId.getBatchSize());
	bitSet.clear(0, batchMessageId.getBatchIndex() + 1);
	return doCumulativeAck(batchMessageId, null, bitSet);

If messageId.equals(this.messageId), it must means this.messageId has the same batch index with messageId. In this case, could bitSetRecyclable be different with this.bitSetRecyclable?

[lordcheng10]

If messageId.equals(this.messageId), it must means this.messageId has the same batch index with messageId.

Yes，you are right!

Can the bitSetRecyclable variable be removed? Because we can know which sub-messages have been acknowledged just by the batch index：@BewareMyPower