mumrah commented on code in PR #13180:
URL: https://github.com/apache/kafka/pull/13180#discussion_r1094683166
##########
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java:
##########
@@ -415,7 +416,7 @@ public void run() throws Exception {
log.info("Migrating {} records from ZK",
batch.size());
}
CompletableFuture<?> future =
zkRecordConsumer.acceptBatch(batch);
- count.addAndGet(batch.size());
+ summary.acceptBatch(batch);
Review Comment:
@clolov thanks for taking a look and welcome to our corner of Kafka 😄
The reason I added the atomic integer here is because
`count.addAndGet(batch.size());` happens inside an anonymous function (lambda).
E.g.,
```java
zkMigrationClient.readAllMetadata(batch -> {
// ...
count.addAndGet(batch.size());
// ...
});
```
Inside lambdas, you can only reference final or effectively final fields
from an outer scope. The `AtomicInteger` reference is effectively final since
we never change the assignment. It's kind of an annoying side-effect of using
lambdas for simple things like for-loops, but it's a good safety measure since
sometimes the lambda can be run in another thread or in parallel (depending on
what you're doing).
With this change, I'm moving the counting into `MigrationSummary`. Since I
know this lambda is executed by a single thread in ZkMigrationClient, I can
safely use a plain `int` to count.
HTH
##########
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java:
##########
@@ -415,7 +416,7 @@ public void run() throws Exception {
log.info("Migrating {} records from ZK",
batch.size());
}
CompletableFuture<?> future =
zkRecordConsumer.acceptBatch(batch);
- count.addAndGet(batch.size());
+ summary.acceptBatch(batch);
Review Comment:
@clolov thanks for taking a look and welcome to our corner of Kafka 😄
The reason I added the atomic integer here is because
`count.addAndGet(batch.size());` happens inside an anonymous function (lambda).
E.g.,
```java
zkMigrationClient.readAllMetadata(batch -> {
// ...
count.addAndGet(batch.size());
// ...
});
```
Inside lambdas, you can only reference final or effectively final fields
from an outer scope. The `AtomicInteger` reference is effectively final since
we never change the assignment. It's kind of an annoying side-effect of using
lambdas for simple things like for-loops, but it's a good safety measure since
sometimes the lambda can be run in another thread or in parallel (depending on
what you're doing).
With this change, I'm moving the counting into `MigrationSummary`. Since I
know this lambda is executed by a single thread in ZkMigrationClient, I can
safely use a plain `int` to count.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
