[GitHub] [flink] RamanVerma commented on a diff in pull request #22150: [FLINK-31305] Propagate producer exceptions outside of mailbox execut…

["RamanVerma (via GitHub)" <gi...@apache.org>]

RamanVerma commented on code in PR #22150:
URL: https://github.com/apache/flink/pull/22150#discussion_r1136403399


##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -397,36 +409,57 @@ private void registerMetricSync() {
                 });
     }
 
+    /** This logic needs to be invoked by write AND flush since we support various semantics. */
+    private void checkAsyncException() throws IOException {
+        // reset this exception since we could close the writer later on
+        Exception e = asyncProducerException.getAndUpdate(ignored -> null);
+        if (e != null) {
+            throw new IOException(
+                    "One or more Kafka Producer send requests have encountered exception", e);
+        }
+    }
+
     private class WriterCallback implements Callback {
         private final MailboxExecutor mailboxExecutor;
         @Nullable private final Consumer<RecordMetadata> metadataConsumer;
 
+        private final AtomicReference<Exception> asyncException;
+
         public WriterCallback(
                 MailboxExecutor mailboxExecutor,
-                @Nullable Consumer<RecordMetadata> metadataConsumer) {
+                @Nullable Consumer<RecordMetadata> metadataConsumer,
+                AtomicReference<Exception> asyncException) {
             this.mailboxExecutor = mailboxExecutor;
             this.metadataConsumer = metadataConsumer;
+            this.asyncException = asyncException;
         }
 
         @Override
         public void onCompletion(RecordMetadata metadata, Exception exception) {
             if (exception != null) {
                 FlinkKafkaInternalProducer<byte[], byte[]> producer =
                         KafkaWriter.this.currentProducer;
-                mailboxExecutor.execute(
-                        () -> {
-                            numRecordsOutErrorsCounter.inc();
-                            throwException(metadata, exception, producer);
-                        },
-                        "Failed to send data to Kafka");
+
+                // Propagate the first exception since amount of exceptions could be large. Need to
+                // do this in Producer IO thread since flush() guarantees that the future will
+                // complete. The same guarantee does not hold for tasks executed in separate
+                // executor e.g. mailbox executor. flush() needs to have the exception immediately
+                // available to fail the checkpoint.
+                asyncException.compareAndSet(
+                        null, decorateException(metadata, exception, producer));
+
+                // Need to send metrics through mailbox thread since we are in the producer io
+                // thread
+                mailboxExecutor.submit(
+                        () -> numRecordsOutErrorsCounter.inc(), "Update error metric");

Review Comment:
   Do we need to throw the exception in mailboxExecutor thread also, as was the case earlier.



##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -142,7 +143,8 @@
         this.deliveryCallback =
                 new WriterCallback(
                         sinkInitContext.getMailboxExecutor(),
-                        sinkInitContext.<RecordMetadata>metadataConsumer().orElse(null));
+                        sinkInitContext.<RecordMetadata>metadataConsumer().orElse(null),

Review Comment:
   Line 155 checks for sinkInitContext to not be null. It should be moved before this line.



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org