[fix][client] Fix unable to send cumulative acknowledgment for last messageId of a batch when enable batch index acknowledgment

[coderzc]

Motivation

When enable batch index acknowledgment, we are unable to send a cumulative acknowledgment for the last messageId of a batch, the root cause is the last messageId of a batch doesn't carry with batch index but due to the wrong comparison in org.apache.pulsar.client.impl.LastCumulativeAck#update make the last messageId can't be sent.

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

Lines 228 to 229 in 545f33f

	if (batchMessageId == null || batchMessageId.ackCumulative()) {
	return doCumulativeAck(msgId, properties, null);

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

Lines 624 to 632 in 545f33f

	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;
	}
	}

    public synchronized void update(final MessageIdImpl messageId, final BitSetRecyclable bitSetRecyclable) {
        if (messageId.compareTo(this.messageId) > 0) {
        // messageId(3,0,-1) < this.messageId(3,0,-1,8) so the following logic cannot be executed
            if (this.bitSetRecyclable != null && this.bitSetRecyclable != bitSetRecyclable) {
                this.bitSetRecyclable.recycle();
            }
            set(messageId, bitSetRecyclable);
            flushRequired = true;
        }
    }

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

Lines 71 to 84 in 82237d3

	public int compareTo(MessageId o) {
	if (o instanceof MessageIdImpl) {
	MessageIdImpl other = (MessageIdImpl) o;
	int batchIndex = (o instanceof BatchMessageIdImpl) ? ((BatchMessageIdImpl) o).batchIndex : NO_BATCH;
	return messageIdCompare(
	this.ledgerId, this.entryId, this.partitionIndex, this.batchIndex,
	other.ledgerId, other.entryId, other.partitionIndex, batchIndex
	);
	} else if (o instanceof TopicMessageIdImpl) {
	return compareTo(((TopicMessageIdImpl) o).getInnerMessageId());
	} else {
	throw new UnsupportedOperationException("Unknown MessageId type: " + o.getClass().getName());
	}
	}

Modifications

Fix the wrong comparison in org.apache.pulsar.client.impl.LastCumulativeAck#update, change it to
(x,y,z) >= (x,y,z,w).

I'm not sure whether modifying MessageImpl.compareTo and BatchMessageIdImpl.compareTo can lead to other problems, so just modify org.apache.pulsar.client.impl.LastCumulativeAck#update

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

testAcknowledgeCumulative

Documentation

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

Matching PR in forked repository

PR in forked repository: coderzc#32

[congbobo184]

LGTM! one question : why not modify the compareTo method directly?

[coderzc]

I'm not sure whether modifying MessageImpl.compareTo and BatchMessageIdImpl.compareTo can lead to other problems, I don't know why the previous logic of compareTo was like that.

[BewareMyPower]

This logic is already included in #18486