[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records

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

yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1214236252


##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##########
@@ -279,30 +280,47 @@ public Future<Void> set(Map<ByteBuffer, ByteBuffer> values, Callback<Void> callb
             throw new IllegalStateException("At least one non-null offset store must be provided");
         }
 
+        boolean containsTombstones = values.entrySet()
+                .stream()
+                .anyMatch(offset -> offset.getValue() == null);
+
         return primaryStore.set(values, (primaryWriteError, ignored) -> {
             if (secondaryStore != null) {
                 if (primaryWriteError != null) {
                     log.trace("Skipping offsets write to secondary store because primary write has failed", primaryWriteError);
+                    try (LoggingContext context = loggingContext()) {
+                        callback.onCompletion(primaryWriteError, ignored);
+                    }
                 } else {
                     try {
                         // Invoke OffsetBackingStore::set but ignore the resulting future; we don't block on writes to this
-                        // backing store.
+                        // backing store. The only exception to this is when a batch consisting of tombstone records fails
+                        // to be written to secondary store and has been successfully written to the primary store. In this case
+                        // an error would be propagated back as in such cases, a deleted source partition
+                        // would be reported as present because the 2 stores are not in sync.
                         secondaryStore.set(values, (secondaryWriteError, ignored2) -> {
                             try (LoggingContext context = loggingContext()) {
-                                if (secondaryWriteError != null) {
+                                if (secondaryWriteError != null && containsTombstones) {
+                                    log.warn("Failed to write offsets with tombstone records to secondary backing store", secondaryWriteError);
+                                    callback.onCompletion(secondaryWriteError, ignored);
+                                    return;
+                                } else if (secondaryWriteError != null) {
                                     log.warn("Failed to write offsets to secondary backing store", secondaryWriteError);
                                 } else {
                                     log.debug("Successfully flushed offsets to secondary backing store");
                                 }
+                                //primaryWriteError is null at this point, and we don't care about secondaryWriteError
+                                callback.onCompletion(null, ignored);

Review Comment:
   Isn't this going to cause us to always block on writes to the secondary store which is something we want to avoid?



-- 
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-unsubscribe@kafka.apache.org

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