[destination-bigquery] Large single-stream sync never receives terminal status, fails with TransientErrorException

### Connector Name

destination-bigquery

### Connector Version

3.0.17

### What step the error happened?

During the sync

### Relevant information

## Environment
- **Airbyte:** 2.0.1
- **Source:** source-mssql 4.3.5 (tested with and without CDC)
- **Destination:** destination-bigquery 3.0.17
- **Table:** Single table ~4.5M rows, ~415 MB

## Error
Sync fails with:
io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform.

## Observed behavior
- BigQuery load jobs complete successfully (`state=DONE`, `badRecords=0`, large `outputRows`).
- No `COMPLETE` (terminal) stream status is ever emitted for this stream, so the Bulk Load CDK fails the sync. Only the first load batch (if any) is committed to the final table; later batches remain in staging.

## Cases tested (all fail with the same error)
| Source config | Sync mode | Result |
|----------------------|------------------------|--------|
| MSSQL with CDC | Full refresh, Overwrite| Fail |
| MSSQL with CDC (after full Reset) | Full refresh, Overwrite | Fail |
| MSSQL without CDC | Full refresh, Overwrite| Fail |
| MSSQL without CDC | Incremental | Fail |

So the issue is **not** specific to CDC vs non-CDC or Full vs Incremental. It reproduces whenever a single large stream is read to completion; the pipeline then reports "input fully read" but never emits a terminal status for that stream.

## Log excerpt (sanitized)
See attached:

[mssql_test_bq_mytable_cdc_full_logs_106_sanitized.txt](https://github.com/user-attachments/files/25954700/mssql_test_bq_mytable_cdc_full_logs_106_sanitized.txt)
[mssql_test_bq_mytable_full_no_cdc_logs_110_sanitized.txt](https://github.com/user-attachments/files/25954701/mssql_test_bq_mytable_full_no_cdc_logs_110_sanitized.txt)

### Relevant log output

```shell
=== LOG 110 EXTRACT (Full refresh, no CDC, single large stream) - SANITIZED ===
Source: MSSQL (no CDC). Destination: BigQuery. Sync: Full refresh | Overwrite.
Table: STREAM_TABLE_NAME (~4.5M rows). Job fails after each BQ load with "streams without terminal status".

>> ATTEMPT 1/3

2026-03-12 22:39:21 info INFO main i.a.c.l.c.DestinationCatalog():49 Destination catalog initialized: [DestinationStream(unmappedNamespace=dbo, unmappedName=STREAM_TABLE_NAME, importType=Append, schema=ObjectType(properties={ID=FieldType(type=StringType, nullable=true), NBRE=FieldType(type=IntegerType, nullable=true), VALEUR=FieldType(type=NumberType, nullable=true), version=FieldType(type=IntegerType, nullable=true), ID_DEVISE=FieldType(type=IntegerType, nullable=true), ID_MONNAIE=FieldType(type=IntegerType, nullable=true), ID_OPERATION_CAIS=FieldType(type=StringType, nullable=true)}, ...), syncId=110, ...)]
2026-03-12 22:39:23 info Sending update for dbo:STREAM_TABLE_NAME - null -> RUNNING
2026-03-12 22:39:23 info Stream Status Update Received: dbo:STREAM_TABLE_NAME - RUNNING
2026-03-12 22:39:30 info Stream status TRACE received of status: STARTED for stream dbo:STREAM_TABLE_NAME
2026-03-12 22:39:32 warn WARN DefaultDispatcher-worker-2#dbo.STREAM_TABLE_NAME-round-1-create-partitions i.a.c.r.JdbcConcurrentPartitionsCreator(run):270 No split boundaries found, using single partition
2026-03-12 22:39:32 info INFO DefaultDispatcher-worker-9 i.a.c.l.s.PipelineEventBookkeepingRouter(handleStreamMessage):101 Saw first record for stream Descriptor(namespace=null, name=STREAM_TABLE_NAME); awaiting setup complete
2026-03-12 22:39:32 info INFO DefaultDispatcher-worker-9 i.a.c.l.s.PipelineEventBookkeepingRouter(handleStreamMessage):105 Setup complete, starting stream Descriptor(namespace=null, name=STREAM_TABLE_NAME)
2026-03-12 22:39:32 info INFO DefaultDispatcher-worker-9 i.a.c.l.o.d.d.DirectLoadTableAppendTruncateStreamLoader(start):145 AppendTruncateStreamLoader starting for stream STREAM_TABLE_NAME
2026-03-12 22:39:32 info INFO DefaultDispatcher-worker-9 i.a.i.d.b.w.t.d.BigqueryDirectLoadNativeTableOperations(ensureSchemaMatches-ifck6jM):82 Stream STREAM_TABLE_NAME has correct schema; no action needed.
2026-03-12 22:39:53 info INFO DefaultDispatcher-worker-9 i.a.c.l.f.SocketInputFlow$collect$2$1(invokeSuspend):44 Read 1000000 messages from /var/run/sockets/airbyte_socket_1.sock
2026-03-12 22:40:07 info INFO DefaultDispatcher-worker-9 i.a.i.d.b.BigQueryUtils(waitForJobFinish):136 Job JobId{project=my-gcp-project, job=99762ede-7177-4c05-a0c9-277c28dee6d5, location=europe-west1} completed successfully, job info Job{... statistics=LoadStatistics{... outputRows=1038639, badRecords=0}, ...}
2026-03-12 22:40:08 info INFO DefaultDispatcher-worker-9 i.a.i.d.b.w.s.BigqueryBatchStandardInsertsLoader(finish):100 Finished loading data into table perfect_demo.STREAM_TABLE_NAME. 1038639 rows loaded; 0 bad records.
2026-03-12 22:40:08 info INFO DefaultDispatcher-worker-9 i.a.c.l.t.i.LoadPipelineStepTask$execute$$inlined$fold$1(emit):317 LoadPipelineStepTask(DirectLoadRecordAccumulator, part=0) saw end-of-stream for Descriptor(namespace=null, name=STREAM_TABLE_NAME) after 1038639 inputs, all workers complete
2026-03-12 22:40:08 error ERROR DefaultDispatcher-worker-9 i.a.c.l.t.DestinationTaskLauncher$WrappedTask(execute):129 Caught exception in task LoadPipelineStepTask(DirectLoadRecordAccumulator, part=0) io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]
at io.airbyte.cdk.load.state.SyncManager.markInputConsumed(SyncManager.kt:116)
at io.airbyte.cdk.load.state.PipelineEventBookkeepingRouter.close(PipelineEventBookkeepingRouter.kt:300)
at io.airbyte.cdk.load.file.SocketInputFlow.collect(SocketInputFlow.kt:93)
at io.airbyte.cdk.load.file.SocketInputFlow$collect$1.invokeSuspend(SocketInputFlow.kt)
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:124)
at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:89)
at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:586)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:820)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704)
2026-03-12 22:40:08 error ERROR main i.a.c.AirbyteConnectorRunnable(run):38 Failed class io.airbyte.cdk.load.write.WriteOperation operation execution. io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]

--- ATTEMPT 2/3 ---

2026-03-12 22:45:06 info INFO DefaultDispatcher-worker-6 i.a.i.d.b.BigQueryUtils(waitForJobFinish):136 Job JobId{project=my-gcp-project, job=6fac5a18-38bb-41cb-82b2-301eba69a75c, location=europe-west1} completed successfully, job info Job{... statistics=LoadStatistics{... outputRows=1610955, badRecords=0}, ...}
2026-03-12 22:45:06 error ERROR DefaultDispatcher-worker-6 i.a.c.l.t.DestinationTaskLauncher$WrappedTask(execute):129 Caught exception in task LoadPipelineStepTask(DirectLoadRecordAccumulator, part=0) io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]
2026-03-12 22:45:06 error ERROR main i.a.c.AirbyteConnectorRunnable(run):38 Failed class io.airbyte.cdk.load.write.WriteOperation operation execution. io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]

--- ATTEMPT 3/3 ---

2026-03-12 22:50:18 info INFO DefaultDispatcher-worker-7 i.a.i.d.b.BigQueryUtils(waitForJobFinish):136 Job JobId{project=my-gcp-project, job=fa15edfb-cc7f-4ed5-9ea6-f742786d9d2e, location=europe-west1} completed successfully, job info Job{... statistics=LoadStatistics{... outputRows=1233383, badRecords=0}, ...}
2026-03-12 22:50:18 error ERROR DefaultDispatcher-worker-7 i.a.c.l.t.DestinationTaskLauncher$WrappedTask(execute):129 Caught exception in task LoadPipelineStepTask(DirectLoadRecordAccumulator, part=0) io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]
2026-03-12 22:50:18 error ERROR main i.a.c.AirbyteConnectorRunnable(run):38 Failed class io.airbyte.cdk.load.write.WriteOperation operation execution. io.airbyte.cdk.TransientErrorException: Input was fully read, but some streams did not receive a terminal stream status message. If the destination did not encounter other errors, this likely indicates an error in the source or platform. Streams without a status message: [STREAM_TABLE_NAME]

=== END EXTRACT ===
Summary: Three attempts. BQ load jobs complete (outputRows=1038639, 1610955, 1233383). Only first batch is committed to final table; sync fails each time with TransientErrorException (no terminal stream status for STREAM_TABLE_NAME).
```

### Contribute

- [x] Yes, I want to contribute

---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/11640