KAFKA-20913: Snapshot CompletedBatch appendTimestamp before buffer reuse

[shantanushukla]

JIRA: https://issues.apache.org/jira/browse/KAFKA-20913

On the Raft leader, KafkaRaftClient.appendBatch writes the batch to the log, registers an async commit callback, then release()s the pooled buffer. CompletedBatch.appendTimestamp() was reading MemoryRecords.firstBatch().maxTimestamp() from a slice of that buffer. Production uses BatchMemoryPool, which requeues the buffer; a later batch overwrites the v2 header. When HWM later completes the callback, handleCommit can deliver the newer batch's append time.

Fix

Snapshot maxTimestamp when CompletedBatch is constructed and return that field from appendTimestamp(). The timestamp is fixed at build time (BatchBuilder writes time.milliseconds() into the header). Record payloads and sizeInBytes() are not affected.

Why existing tests missed this

• RaftClientTestContext defaults to MemoryPool.NONE, which allocates a new buffer every time and never reuses.
• BatchAccumulatorTest called appendTimestamp() before release(), or verified release() without reading the timestamp afterward.
• If the listener is still at the starting offset, commits come from log replay (the log copies records at append time), which hides the in-memory path.

Tests

These failed on trunk (first batch reported T2) and pass after the snapshot:

• BatchAccumulatorTest.testAppendTimestampUnchangedAfterBufferReleasedAndReused — one-buffer BatchMemoryPool; drain/release at T1, append at T2; first batch must still report T1.
• KafkaRaftClientTest.testLeaderHandleCommitAppendTimestampStableWhenPoolReusesBuffer — two voters so HWM lags; catch the listener up past the leader-change batch; append at T1 then T2 with a single reused buffer; after commit the listener's first data batch must still be T1.

Related existing tests still pass: full BatchAccumulatorTest, testListenerCommitCallbackAfterLeaderWrite, testAccumulatorClearedAfterBecomingFollower.

./gradlew :raft:spotlessApply :raft:checkstyleMain :raft:checkstyleTest :raft:spotlessCheck
./gradlew :raft:test --tests org.apache.kafka.raft.internals.BatchAccumulatorTest --tests org.apache.kafka.raft.KafkaRaftClientTest.testLeaderHandleCommitAppendTimestampStableWhenPoolReusesBuffer --tests org.apache.kafka.raft.KafkaRaftClientTest.testListenerCommitCallbackAfterLeaderWrite --tests org.apache.kafka.raft.KafkaRaftClientTest.testAccumulatorClearedAfterBecomingFollower

BUILD SUCCESSFUL.