Bulk load cdk: throw "no end-of-stream" error as transient (#51608)

airbytehq · Jan 17, 2025 · c9e8c81 · c9e8c81
1 parent 787d6fa
commit c9e8c81
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/state/SyncManager.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/state/SyncManager.kt
@@ -5,6 +5,7 @@
 package io.airbyte.cdk.load.state
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings
+import io.airbyte.cdk.TransientErrorException
 import io.airbyte.cdk.load.command.DestinationCatalog
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.write.StreamLoader
@@ -141,7 +142,7 @@ class DefaultSyncManager(
                 .map { (stream, _) -> stream }
         if (incompleteStreams.isNotEmpty()) {
             val prettyStreams = incompleteStreams.map { it.toPrettyString() }
-            throw IllegalStateException(
+            throw TransientErrorException(
                 "Input was fully read, but some streams did not receive a terminal stream status message. This likely indicates an error in the source or platform. Streams without a status message: $prettyStreams"
             )
         }

diff --git a/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/state/SyncManagerTest.kt b/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/state/SyncManagerTest.kt
@@ -4,6 +4,7 @@
 
 package io.airbyte.cdk.load.state
 
+import io.airbyte.cdk.TransientErrorException
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.command.MockDestinationCatalogFactory.Companion.stream1
 import io.airbyte.cdk.load.command.MockDestinationCatalogFactory.Companion.stream2
@@ -133,7 +134,7 @@ class SyncManagerTest {
         val manager1 = syncManager.getStreamManager(stream1.descriptor)
         manager1.markEndOfStream(true)
         // This should fail, because stream2 was not marked with end of stream
-        val e = assertThrows<IllegalStateException> { syncManager.markInputConsumed() }
+        val e = assertThrows<TransientErrorException> { syncManager.markInputConsumed() }
         assertEquals(
             // stream1 is fine, so the message only includes stream2
             "Input was fully read, but some streams did not receive a terminal stream status message. This likely indicates an error in the source or platform. Streams without a status message: [test.stream2]",