[GitHub] [kafka] C0urante commented on a change in pull request #11524: KAFKA-13469: Block for in-flight record delivery before end-of-life source task offset commit

[GitBox]

C0urante commented on a change in pull request #11524:
URL: https://github.com/apache/kafka/pull/11524#discussion_r755307964



##########
File path: 
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SubmittedRecords.java
##########
@@ -68,6 +75,9 @@ SubmittedRecord submit(Map<String, Object> partition, 
Map<String, Object> offset
         SubmittedRecord result = new SubmittedRecord(partition, offset);
         records.computeIfAbsent(result.partition(), p -> new LinkedList<>())
                 .add(result);
+        synchronized (this) {
+            numUnackedMessages.incrementAndGet();
+        }

Review comment:
       It definitely feels like an anti-pattern to synchronize around 
modifications to an atomic field. I've gone into further detail below but 
TL;DR: this is a combination of wanting to avoid a race condition (hence the 
synchronized block) and wanting to silence SpotBugs warnings (hence the use of 
an `AtomicInteger` instead of a `volatile int`).




-- 
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