mumrah commented on code in PR #15007:
URL: https://github.com/apache/kafka/pull/15007#discussion_r1427480307
##########
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java:
##########
@@ -645,6 +648,29 @@ public void run() throws Exception {
}
}
+ private BufferingBatchConsumer<ApiMessageAndVersion>
buildMigrationBatchConsumer(
+ MigrationManifest.Builder manifestBuilder
+ ) {
+ return new BufferingBatchConsumer<>(batch -> {
+ try {
+ if (log.isTraceEnabled()) {
+ batch.forEach(apiMessageAndVersion ->
+
log.trace(recordRedactor.toLoggableString(apiMessageAndVersion.message())));
+ }
+ CompletableFuture<?> future =
zkRecordConsumer.acceptBatch(batch);
Review Comment:
To answer your question more directly
> Does the consumer here have any expectation on atomicity of the records?
No. The eventual consumer of these batches is
QuorumController#MigrationRecordConsumer which simply sends them along to Raft
as a non-atomic batch. It doesn't care about batch boundaries or alignment.
--
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]
