niket-goel commented on a change in pull request #10899:
URL: https://github.com/apache/kafka/pull/10899#discussion_r660135431
##########
File path:
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -204,35 +207,103 @@ private void completeCurrentBatch() {
currentBatch = null;
}
- public void appendLeaderChangeMessage(LeaderChangeMessage
leaderChangeMessage, long currentTimeMs) {
+ private void appendControlMessage(
+ Supplier<MemoryRecords> supplier,
+ ByteBuffer buffer
+ ) {
appendLock.lock();
try {
forceDrain();
- ByteBuffer buffer = memoryPool.tryAllocate(256);
- if (buffer != null) {
- MemoryRecords data = MemoryRecords.withLeaderChangeMessage(
- this.nextOffset,
- currentTimeMs,
- this.epoch,
- buffer,
- leaderChangeMessage
- );
- completed.add(new CompletedBatch<>(
- nextOffset,
- 1,
- data,
- memoryPool,
- buffer
- ));
- nextOffset += 1;
- } else {
- throw new IllegalStateException("Could not allocate buffer for
the leader change record.");
- }
+ completed.add(new CompletedBatch<>(
+ nextOffset,
+ 1,
+ supplier.get(),
+ memoryPool,
+ buffer
+ ));
+ nextOffset += 1;
} finally {
appendLock.unlock();
}
}
+ /**
+ * Append a {@link LeaderChangeMessage} record to the batch
+ *
+ * @param @LeaderChangeMessage The message to append
+ * @param @currentTimeMs The timestamp of message generation
+ * @throws IllegalStateException on failure to allocate a buffer for the
record
+ */
+ public void appendLeaderChangeMessage(
+ LeaderChangeMessage leaderChangeMessage,
+ long currentTimeMs
+ ) {
+ ByteBuffer buffer = memoryPool.tryAllocate(256);
Review comment:
The code organization gets funny if we do that because the byte buffer
is needed for the MemoryRecord API as well. Will sync with you offline on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
