demo: do not review | CC-41637: Commit the frontier offset in the S3 sink to clear phantom lag and restart re-reads

kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java

@@ -200,6 +200,20 @@ public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
       + " is configured";
   public static final boolean ROTATE_FILE_ON_PARTITION_CHANGE_DEFAULT = true;
 
+  public static final String COMMIT_FRONTIER_OFFSET_CONFIG = "commit.frontier.offset.enable";
+  public static final boolean COMMIT_FRONTIER_OFFSET_DEFAULT = false;
+  public static final String COMMIT_FRONTIER_OFFSET_DOC =
+      "When enabled, preCommit reports the frontier offset per partition instead of only the offset"
+      + " of the last uploaded file. The frontier is the lowest Kafka offset of a record that is"
+      + " buffered but not yet uploaded to S3, or the consumed high-water mark when nothing is"
+      + " buffered. Everything below the frontier is decided (already uploaded, or dropped or"
+      + " filtered), so committing it clears the phantom lag that a partition otherwise shows when"
+      + " records are consumed but no file has rotated yet (idle partitions, runs of filtered or"
+      + " null-valued records, or buffers still under the flush size). Because S3 resumes from the"
+      + " committed Kafka offset on restart, committing the frontier never steps over a record that"
+      + " is still in flight. Defaults to false, which preserves the existing behavior.";
+  public static final String COMMIT_FRONTIER_OFFSET_DISPLAY = "Commit Frontier Offset";
+
   /**
    * Maximum back-off time when retrying failed requests.
    */
@@ -759,6 +773,18 @@ public static ConfigDef newConfigDef() {
           "Rotate files on partition change"
       );
 
+      configDef.define(
+          COMMIT_FRONTIER_OFFSET_CONFIG,
+          Type.BOOLEAN,
+          COMMIT_FRONTIER_OFFSET_DEFAULT,
+          Importance.LOW,
+          COMMIT_FRONTIER_OFFSET_DOC,
+          group,
+          ++orderInGroup,
+          Width.SHORT,
+          COMMIT_FRONTIER_OFFSET_DISPLAY
+      );
+
       // This is done to avoid aggressive schema based rotations resulting out of interleaving
       // of tombstones with regular records.
       configDef.define(
@@ -1418,6 +1444,10 @@ public boolean shouldRotateOnPartitionChange() {
     return getBoolean(ROTATE_FILE_ON_PARTITION_CHANGE);
   }
 
+  public boolean isCommitFrontierOffsetEnabled() {
+    return getBoolean(COMMIT_FRONTIER_OFFSET_CONFIG);
+  }
+
   public enum IgnoreOrFailBehavior {
     IGNORE,
     FAIL;

kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkTask.java

@@ -352,6 +352,9 @@ public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
   public Map<TopicPartition, OffsetAndMetadata> preCommit(
       Map<TopicPartition, OffsetAndMetadata> offsets
   ) {
+    if (connectorConfig.isCommitFrontierOffsetEnabled()) {
+      return preCommitFrontier(offsets);
+    }
     Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
     for (TopicPartition tp : topicPartitionWriters.keySet()) {
       Long offset = topicPartitionWriters.get(tp).getOffsetToCommitAndReset();
@@ -363,6 +366,37 @@ public Map<TopicPartition, OffsetAndMetadata> preCommit(
     return offsetsToCommit;
   }
 
+  /**
+   * Computes the offset to commit per partition as the frontier rather than the offset of the last
+   * uploaded file. The frontier is the lowest Kafka offset of a record that is buffered but not
+   * yet uploaded to S3, or the consumed high-water mark from the {@code offsets} argument when
+   * nothing is buffered. Everything below the frontier is decided, so committing it clears the
+   * phantom lag that a partition otherwise carries when records are consumed but no file has
+   * rotated (idle partitions, runs of filtered or null-valued records, or buffers still under the
+   * flush size). Because S3 resumes from the committed Kafka offset on restart, committing the
+   * frontier never steps over an in-flight record.
+   */
+  private Map<TopicPartition, OffsetAndMetadata> preCommitFrontier(
+      Map<TopicPartition, OffsetAndMetadata> offsets
+  ) {
+    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
+    for (TopicPartition tp : topicPartitionWriters.keySet()) {
+      // The framework's consumed position includes offsets that never reached put(), such as
+      // SMT-filtered records, so it is the correct upper bound for a partition with nothing
+      // buffered. It may be absent for a partition the framework has not reported yet.
+      OffsetAndMetadata consumed = offsets.get(tp);
+      Long consumedHighWaterMark = consumed != null ? consumed.offset() : null;
+      Long frontier =
+          topicPartitionWriters.get(tp).getFrontierOffset(consumedHighWaterMark);
+      if (frontier != null) {
+        log.trace("Forwarding to framework request to commit frontier offset: {} for {}",
+            frontier, tp);
+        offsetsToCommit.put(tp, new OffsetAndMetadata(frontier));
+      }
+    }
+    return offsetsToCommit;
+  }
+
   @Override
   public void close(Collection<TopicPartition> partitions) {
     for (TopicPartition tp : topicPartitionWriters.keySet()) {

kafka-connect-s3/src/main/java/io/confluent/connect/s3/TopicPartitionWriter.java

@@ -53,6 +53,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
+import java.util.TreeSet;
 
 import io.confluent.common.utils.SystemTime;
 import io.confluent.common.utils.Time;
@@ -104,6 +105,13 @@ public class TopicPartitionWriter {
   private Long offsetToCommit;
   private final RecordWriterProvider<S3SinkConnectorConfig> writerProvider;
 
+  // The set P of pending offsets: records that have been buffered for this partition but are not
+  // yet durable in S3. An offset is added when the record is buffered and removed once the file
+  // holding it has been committed. The frontier is min(P) when P is non-empty, so this stays
+  // ordered for a cheap first() lookup.
+  private final boolean commitFrontierOffset;
+  private final TreeSet<Long> pendingOffsets = new TreeSet<>();
+
   // VisibleForTesting
   Map<Long, String> offsetToFilenameMap;
 
@@ -204,6 +212,7 @@ public TopicPartitionWriter(TopicPartition tp,
         connectorConfig.getString(StorageSinkConnectorConfig.SCHEMA_COMPATIBILITY_CONFIG));
 
     buffer = new LinkedList<>();
+    commitFrontierOffset = connectorConfig.isCommitFrontierOffsetEnabled();
     commitFiles = new LinkedHashMap<>();
     writers = new HashMap<>();
     currentSchemas = new HashMap<>();
@@ -302,7 +311,9 @@ public void write() {
         throw new ConnectException(e);
       } catch (SchemaProjectorException e) {
         if (reporter != null) {
-          reporter.report(buffer.poll(), e);
+          SinkRecord errantRecord = buffer.poll();
+          markDecidedOffset(errantRecord);
+          reporter.report(errantRecord, e);
           log.warn("Errant record written to DLQ due to: {}", e.getMessage());
         } else {
           throw e;
@@ -366,7 +377,7 @@ private void executeState(long now) {
         } catch (PartitionException e) {
           if (reporter != null) {
             reporter.report(record, e);
-            buffer.poll();
+            markDecidedOffset(buffer.poll());
             break;
           } else {
             throw e;
@@ -504,8 +515,11 @@ private boolean checkRotationOrAppend(
 
     SinkRecord projectedRecord = compatibility.project(record, null, currentValueSchema);
     boolean validRecord = writeRecord(projectedRecord, encodedPartition, record);
-    buffer.poll();
+    SinkRecord bufferedRecord = buffer.poll();
     if (!validRecord) {
+      // The record was faulty and went to the DLQ, so it is decided and leaves the pending set.
+      // It stays in the set on the valid path until the file holding it is committed.
+      markDecidedOffset(bufferedRecord);
       // skip the faulty record and don't rotate
       return false;
     }
@@ -585,6 +599,23 @@ public void close() throws ConnectException {
 
   public void buffer(SinkRecord sinkRecord) {
     buffer.add(sinkRecord);
+    if (commitFrontierOffset) {
+      // Record the offset in the pending set the moment it is buffered, and before put() returns
+      // to the framework. This is the single place a record enters P, so no code path can accept a
+      // record without it becoming visible to the frontier computation.
+      pendingOffsets.add(sinkRecord.kafkaOffset());
+    }
+  }
+
+  /**
+   * Marks a buffered record as decided without uploading it, which happens when it is sent to the
+   * dead-letter queue or dropped by a partition error. The record will never be uploaded, so it
+   * leaves the pending set and stops holding the frontier back.
+   */
+  private void markDecidedOffset(SinkRecord sinkRecord) {
+    if (commitFrontierOffset && sinkRecord != null) {
+      pendingOffsets.remove(sinkRecord.kafkaOffset());
+    }
   }
 
   public Long getOffsetToCommitAndReset() {
@@ -593,6 +624,23 @@ public Long getOffsetToCommitAndReset() {
     return latest;
   }
 
+  /**
+   * Returns the frontier offset to commit for this partition: the lowest offset of a record that
+   * is buffered but not yet uploaded, or the consumed high-water mark when nothing is buffered.
+   * Every offset below the frontier is decided, either already uploaded to S3, or dropped or
+   * filtered before it could be buffered, so committing it never steps over an in-flight record.
+   *
+   * @param consumedHighWaterMark the offset one past the last record the framework has handed to
+   *                              put() for this partition, as supplied by preCommit
+   * @return the frontier offset, or null when this partition has no consumed position to report
+   */
+  public Long getFrontierOffset(Long consumedHighWaterMark) {
+    if (!pendingOffsets.isEmpty()) {
+      return pendingOffsets.first();
+    }
+    return consumedHighWaterMark;
+  }
+
   public Long currentStartOffset() {
     return minStartOffset();
   }
@@ -976,6 +1024,12 @@ private void commitFiles() {
     }
 
     offsetToCommit = currentOffset + 1;
+    if (commitFrontierOffset) {
+      // Every offset at or below currentOffset has now been uploaded to S3 and is durable, so it
+      // leaves the pending set. The frontier then advances to the next record still in flight, or
+      // to the consumed high-water mark once nothing is left buffered.
+      pendingOffsets.headSet(currentOffset, true).clear();
+    }
     commitFiles.clear();
     currentSchemas.clear();
     offsetToFilenameMap.clear();