feat: connection manager workflow considers rejected records for progress (#16914)

Author: gosusnp

airbyte-workers/src/main/java/io/airbyte/workers/helpers/ProgressCheckerPredicates.java

@@ -27,8 +27,16 @@ public boolean test(@Nonnull final AttemptStats stats) {
     Objects.requireNonNull(stats, "Attempt stats null. Cannot test progress.");
     // If recordsCommitted is null, treat this as a 0
     final var recordsCommitted = Objects.requireNonNullElse(stats.getRecordsCommitted(), 0L);
+    final var recordsRejected = Objects.requireNonNullElse(stats.getRecordsRejected(), 0L);
 
-    return recordsCommitted >= RECORDS_COMMITTED_THRESHOLD;
+    // In the context of progress, we count both committed and rejected records since rejected records
+    // are
+    // records that were successfully sent to the destination and rejected by the destination due to the
+    // records
+    // not confirming to the destination requirements. They are being tracked as part of checkpointing
+    // and we should
+    // be continuing past those records to avoid getting stuck on those.
+    return recordsCommitted + recordsRejected >= RECORDS_COMMITTED_THRESHOLD;
   }
 
 }

airbyte-workers/src/test/java/io/airbyte/workers/helpers/ProgressCheckerPredicatesTest.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2020-2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.workers.helpers
+
+import io.airbyte.api.client.model.generated.AttemptStats
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
+import java.util.stream.Stream
+
+internal class ProgressCheckerPredicatesTest {
+  val checker: ProgressCheckerPredicates = ProgressCheckerPredicates()
+
+  @ParameterizedTest
+  @MethodSource("statsProgressMatrix")
+  fun checkAttemptStats(
+    stats: AttemptStats?,
+    expected: Boolean,
+  ) {
+    Assertions.assertEquals(expected, checker.test(stats))
+  }
+
+  companion object {
+    @JvmStatic
+    fun statsProgressMatrix(): Stream<Arguments> =
+      Stream.of(
+        Arguments.of(AttemptStats(recordsCommitted = 0L), false),
+        Arguments.of(AttemptStats(recordsCommitted = 1L), true),
+        Arguments.of(AttemptStats(recordsCommitted = 3L), true),
+        Arguments.of(AttemptStats(recordsCommitted = 3L, recordsRejected = 2L), true),
+        Arguments.of(AttemptStats(recordsRejected = 5L), true),
+        Arguments.of(AttemptStats(recordsCommitted = 9999L), true),
+        Arguments.of(AttemptStats(recordsCommitted = null), false),
+      )
+  }
+}