[fix][client] Keep transactional and non-transactional messages in separate batches - #26547

Motivation

A batch carries a single transaction id in its metadata, so every message inside it inherits that
transaction — there is no per-message override:

// BatchMessageContainerImpl.createOpSendMsg
if (currentTxnidMostBits != -1) {
    messageMetadata.setTxnidMostBits(currentTxnidMostBits);
}

hasSameTxn, the admission check meant to enforce that invariant, cannot express it:

public boolean hasSameTxn(MessageImpl<?> msg) {
    if (!msg.getMessageBuilder().hasTxnidMostBits() || !msg.getMessageBuilder().hasTxnidLeastBits()) {
        return true;                                    // (1)
    }
    if (currentTxnidMostBits == -1 || currentTxnidLeastBits == -1) {
        currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits();   // (2) mutates the batch
        currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
        return true;
    }
    return currentTxnidMostBits == msg.getMessageBuilder().getTxnidMostBits()   // (3)
            && currentTxnidLeastBits == msg.getMessageBuilder().getTxnidLeastBits();
}

Branch (1) inspects only the message, never the batch, so a plain message is admitted into any batch
including a transactional one. Branch (2) performs a write inside a hasSameXxx query, and makes a
plain batch adopt the transaction of the first transactional message merely offered to it. Only
branch (3) actually compares, and it is reachable only when both sides carry a transaction. The check
can answer "are these two transactions the same?" but never "is one of them absent?".

Three consequences, all reachable from a single batching producer used for both newMessage(txn) and
newMessage() — legal API usage, with batching enabled by default:

1. A plain message batched after transactional ones inherits the transaction: it stays invisible until
   commit and is silently discarded if the transaction aborts, although the application never sent
   it inside a transaction.
2. A transactional message joining a batch that already holds plain messages stamps its transaction
   onto the whole batch, retroactively enrolling those plain messages in it.
3. The side effect in branch (2) also fires on the duplicate-sequence-id path in ProducerImpl:
   canAddToCurrentBatch adopts the transaction, and the caller then flushes the current batch via
   doBatchSendAndAdd and puts the new message into a fresh one — so a batch of plain messages is
   published stamped with a transaction that none of its messages belongs to.

BatchMessageKeyBasedContainer fails the other way round. The guard reads the outer container's
transaction state, while the batch metadata is built by the per-key inner container from its own
first message. A transactional message landing in a key bucket whose first message was plain is
therefore published with no transaction id at all — outside the transaction, and not rolled back when
it aborts.

For reference, the sibling admission check on the same line of canAddToCurrentBatch is already the
correct shape: hasSameSchema is a pure query that starts with numMessagesInBatch == 0 and treats
"absent" as an identity that must match.

Modifications

• AbstractBatchMessageContainer.hasSameTxn is now a pure query comparing transaction identities, with
  "no transaction" as an identity of its own that is incompatible with any transaction. An empty batch
  (numMessagesInBatch == 0) still accepts anything; testing the message count rather than the -1
  sentinel is what distinguishes "no identity yet" from "identity is no transaction". Adopting the
  identity is left solely to add, where it already happened — which also removes consequence 3 above.
• BatchMessageKeyBasedContainer.add captures the transaction id from the container's first message.
  This is required rather than cosmetic: with the side effect gone, the outer container's transaction
  state would otherwise stay unset forever, batchHasTxn would always be false, and plain messages
  would again be admitted into a transactional container.

Nothing changes at the call site. When the guard rejects a message, ProducerImpl.doBatchSendAndAdd
already flushes the current batch and opens a new one with that message, exactly as it does today when
the batch is full or the schema differs.

Applications that mix both kinds of send on one producer will see batches split more often. That is the
necessary cost: the two cannot share a batch without one of them inheriting the wrong transaction
identity. There is no wire format or API change, and no behaviour change for applications that do not
mix.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added BatchMessageContainerImplTest.testPlainMessageIsNotBatchedWithTransactionalMessages and
  ...testTransactionalMessageIsNotBatchedWithPlainMessages, covering both directions, and
  ...testKeyBasedContainerKeepsTransactionalAndPlainMessagesApart for the key-based batcher. All
  three fail before the change.
• Added BatchMessageContainerImplTest.testSameTransactionStillBatchesTogether and
  ...testPlainMessagesStillBatchTogether as guards that messages of one transaction still share a
  batch, that two different transactions still do not, and that plain messages still batch together.
• Added TransactionProduceTest.testAbortedTransactionDoesNotDiscardPlainMessageBatchedWithIt: a
  batching producer sends a transactional and a plain message into the same batch, the transaction is
  aborted, and the plain message must still be readable. It fails before the change with "the plain
  message was discarded by the aborted transaction".
• BatchMessageContainerImplTest passes 9/9, the pulsar-client module suite shows no new failures,
  and ./gradlew quickCheck is clean.

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