fix(bulk cdk): cdc termination condition (#76054)

Author: mwbayley

airbyte-cdk/bulk/core/extract/changelog.md

@@ -9,6 +9,7 @@ The Extract CDK provides functionality for source connectors including schema di
 
 | Version | Date       | Pull Request                                              | Subject                                                                                                                                              |
 |---------|------------|-----------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 1.1.1   | 2026-04-02 | [76054](https://github.com/airbytehq/airbyte/pull/76054)  | Clarified CDC termination condition. Ensure querySingleValue returns one result or throws.                                                           |
 | 1.1.0   | 2026-03-30 | [75636](https://github.com/airbytehq/airbyte/pull/75636)  | CDC positions are partially ordered. Added optional CDC startup hook. Field is now EmittedField.                                                     |
 | 1.0.2   | 2026-02-24 | [74007](https://github.com/airbytehq/airbyte/pull/74007)  | Fix infinite loop when cursor-based incremental sync encounters an empty table with prior state by caching NULL cursor upper bound values (toolkit). |
 | 1.0.1   | 2026-02-24 |                                                           | Bump bulk-cdk-core-base to 1.0.1 to pick up CVE fixes (CVE-2021-47621, CVE-2022-36944).                                                              |

airbyte-cdk/bulk/core/extract/version.properties

@@ -1 +1 @@
-version=1.1.0
+version=1.1.1

airbyte-cdk/bulk/toolkits/extract-cdc/src/main/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionReader.kt

@@ -435,20 +435,21 @@ class CdcPartitionReader<T : PartiallyOrdered<T>>(
                 }
             }
 
-            if (upperBound.isGreater(currentPosition)) {
-                return null
-            }
-            // Close because the current event is past the sync upper bound.
-            return when (eventType) {
-                EventType.TOMBSTONE,
-                EventType.HEARTBEAT -> CloseReason.HEARTBEAT_OR_TOMBSTONE_REACHED_TARGET_POSITION
-                EventType.KEY_JSON_INVALID,
-                EventType.VALUE_JSON_INVALID,
-                EventType.RECORD_EMITTED,
-                EventType.RECORD_DISCARDED_BY_DESERIALIZE,
-                EventType.RECORD_DISCARDED_BY_STREAM_ID ->
-                    CloseReason.RECORD_REACHED_TARGET_POSITION
+            if (currentPosition.isGreaterOrEqual(upperBound)) {
+                // Close because the current event is at or past the sync upper bound.
+                return when (eventType) {
+                    EventType.TOMBSTONE,
+                    EventType.HEARTBEAT ->
+                        CloseReason.HEARTBEAT_OR_TOMBSTONE_REACHED_TARGET_POSITION
+                    EventType.KEY_JSON_INVALID,
+                    EventType.VALUE_JSON_INVALID,
+                    EventType.RECORD_EMITTED,
+                    EventType.RECORD_DISCARDED_BY_DESERIALIZE,
+                    EventType.RECORD_DISCARDED_BY_STREAM_ID ->
+                        CloseReason.RECORD_REACHED_TARGET_POSITION
+                }
             }
+            return null // Keep processing.
         }
 
         private fun position(sourceRecord: SourceRecord?): T? {

airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/read/SelectQuerier.kt

@@ -204,6 +204,13 @@ class JdbcSelectQuerier(
     }
 }
 
+private val throwNoResultsIllegalState = {
+    throw IllegalStateException("Query unexpectedly produced no results")
+}
+private val throwMultipleResultsIllegalState = {
+    throw IllegalStateException("Query unexpectedly produced multiple results")
+}
+
 /**
  * Convenience function for executing a query that is expected to return exactly one row and
  * extracting a single value from that row.
@@ -222,19 +229,23 @@ fun <T> querySingleValue(
     query: String,
     bindParameters: ((PreparedStatement) -> Unit)? = null,
     withResultSet: (ResultSet) -> T,
-    noResultsCase: () -> Unit = {
-        throw IllegalStateException("Query unexpectedly produced no results: [$query]")
-    },
-    multipleResultsCase: () -> Unit = {
-        throw IllegalStateException("Query unexpectedly produced more than one result: [$query]")
-    },
+    noResultsCase: () -> Unit = throwNoResultsIllegalState,
+    multipleResultsCase: () -> Unit = throwMultipleResultsIllegalState,
 ): T {
     jdbcConnectionFactory.get().use { connection ->
         connection.prepareStatement(query).use { stmt ->
             bindParameters?.invoke(stmt)
             stmt.executeQuery().use { rs ->
-                if (!rs.next()) noResultsCase()
-                if (!rs.isLast) multipleResultsCase()
+                if (!rs.next()) {
+                    noResultsCase()
+                    // if non-throwing noResultsCase was supplied, we still need to throw
+                    throwNoResultsIllegalState()
+                }
+                if (!rs.isLast) {
+                    multipleResultsCase()
+                    // if non-throwing multipleResultsCase was supplied, we still need to throw
+                    throwMultipleResultsIllegalState()
+                }
                 return withResultSet(rs)
             }
         }