[fix][txn] Ack all message ids when ack chunk messages with transaction.

[liangyepianzhou]

Motivation

Now, only the last chunk will be acknowledged when acknowledging chunk messages with transactions.
If the messageId is a ChunkMessageIdImpl, the ledger ID and entry ID will belong to the lastChunkMsgId.

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

Lines 2791 to 2814 in 2b5c199

	private CompletableFuture<Void> doTransactionAcknowledgeForResponse(MessageId messageId, AckType ackType,
	ValidationError validationError,
	Map<String, Long> properties, TxnID txnID) {
	long requestId = client.newRequestId();
	final MessageIdAdv messageIdAdv = (MessageIdAdv) messageId;
	final long ledgerId = messageIdAdv.getLedgerId();
	final long entryId = messageIdAdv.getEntryId();
	final ByteBuf cmd;
	if (MessageIdAdvUtils.isBatch(messageIdAdv)) {
	BitSetRecyclable bitSetRecyclable = BitSetRecyclable.create();
	bitSetRecyclable.set(0, messageIdAdv.getBatchSize());
	if (ackType == AckType.Cumulative) {
	MessageIdAdvUtils.acknowledge(messageIdAdv, false);
	bitSetRecyclable.clear(0, messageIdAdv.getBatchIndex() + 1);
	} else {
	bitSetRecyclable.clear(messageIdAdv.getBatchIndex());
	}
	cmd = Commands.newAck(consumerId, ledgerId, entryId, bitSetRecyclable, ackType, validationError, properties,
	txnID.getLeastSigBits(), txnID.getMostSigBits(), requestId, messageIdAdv.getBatchSize());
	bitSetRecyclable.recycle();
	} else {
	cmd = Commands.newAck(consumerId, ledgerId, entryId, null, ackType, validationError, properties,
	txnID.getLeastSigBits(), txnID.getMostSigBits(), requestId);
	}

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

Lines 30 to 33 in 2b5c199

	public ChunkMessageIdImpl(MessageIdImpl firstChunkMsgId, MessageIdImpl lastChunkMsgId) {
	super(lastChunkMsgId.getLedgerId(), lastChunkMsgId.getEntryId(), lastChunkMsgId.getPartitionIndex());
	this.firstChunkMsgId = firstChunkMsgId;
	}

Modifications

Flow the common message acknowledge logic, ack all the chunks when acknowledging messages with transactions.

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.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository: