@kapa.ai This error started happening in the connector: Warning from source: The sync encountered an unexpected error in the change event producer and has stopped. Please check the logs for details and troubleshoot accordingly.

io.airbyte.cdk.integrations.source.relationaldb.state.FailedRecordIteratorException: java.lang.RuntimeException: java.lang.RuntimeException: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.

The knowledge sources show this exact pattern of error coming from Debezium-based CDC sources (Postgres/MySQL/MongoDB v2, etc.) when the embedded Debezium engine fails and propagates as:

FailedRecordIteratorException → RuntimeException → org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.

[[MySQL CDC issue](https://github.com/airbytehq/airbyte/issues/50856); [MongoDB v2 CDC issue](https://github.com/airbytehq/airbyte/issues/63300)] However, they do not provide a single root cause or a generic fix; in each case the underlying reason is specific (e.g., invalid configuration, binlog/replica settings, parsing issues). So I can’t tell you the exact cause in your environment from these documents alone. From the available information, the next concrete steps are: 1. Inspect full connector logs around the failure time The actual cause is usually a more specific Debezium or database error a few lines above the

ConnectException

(e.g., invalid configuration, binlog/LSN issue, replica set missing, parsing error). [[MySQL CDC issue](https://github.com/airbytehq/airbyte/issues/50856); [Postgres offset error](https://discuss.airbyte.io/t/2793)] 2. Look for configuration-related errors - For MongoDB v2, a common cause is invalid CDC configuration such as missing

replicaSet

in the connection string for a replica set cluster, which leads to the same

FailedRecordIteratorException

pattern. [[MongoDB v2 CDC issue](https://github.com/airbytehq/airbyte/issues/63300)] - For MySQL/Postgres, similar Debezium errors can be triggered by binlog/LSN or schema parsing problems. [[MySQL CDC issue](https://github.com/airbytehq/airbyte/issues/50856); [Postgres offset error](https://discuss.airbyte.io/t/2793)] 3. Check CDC-specific settings on the source DB The examples show that misconfigured or stressed source databases (binlog settings, WAL/LSN, resources) can cause Debezium to stop with this producer error. [[Postgres offset error](https://discuss.airbyte.io/t/2793); [MySQL CDC issue](https://github.com/airbytehq/airbyte/issues/50856)] 4. Try a newer connector version or adjust advanced CDC behavior - In some Postgres CDC cases, upgrading the connector or changing the “Invalid CDC position behavior” to force a re-sync helped recover from bad CDC state (at the cost of an initial full refresh). [[CDC position behavior](https://github.com/airbytehq/airbyte/issues/55846); [Postgres offset error](https://discuss.airbyte.io/t/2793)] Because the knowledge sources don’t include your exact connector type, version, or the detailed log lines above the

ConnectException

, I can’t narrow it further. To proceed, you’ll need to: - Open the job logs for the failing sync. - Find the first Debezium / database error before the

ConnectException

. - Use that specific message (e.g., configuration invalid, binlog parse error, missing replicaSet, etc.) to guide the fix. If you paste that earlier part of the log (a few dozen lines before the

ConnectException

), I can help interpret it more precisely based only on the documented patterns.

@kapa.ai OK. The source is a Postgres database using CDC. the source connector version is: 3.7.0 The BigQuery destination connector version is: 3.0.10 Here are some more logs message around the event:

2026-02-12 07:06:22 destination 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: [dttp_providers, nationalities, bulk_update_trainee_uploads, degrees, funding_method_subjects, hesa_metadata, funding_methods, nationalisations, apply_applications, courses, withdrawal_reasons, allocation_subjects, bulk_update_row_errors, users, schools, training_partners, subject_specialisms, trainees, placements, academic_cycles, trainee_withdrawal_reasons, trs_trn_requests, course_subjects, potential_duplicate_trainees, dttp_schools, subjects, disabilities, trainee_withdrawals, activities, provider_users, sign_offs, training_partner_users, providers]
	at io.airbyte.cdk.load.state.SyncManager.markInputConsumed(SyncManager.kt:116) ~[bulk-cdk-core-load-0.1.54.jar:?]
	at io.airbyte.cdk.load.state.PipelineEventBookkeepingRouter.close(PipelineEventBookkeepingRouter.kt:300) ~[bulk-cdk-core-load-0.1.54.jar:?]
	at io.airbyte.cdk.load.task.internal.InputConsumerTask.execute(InputConsumerTask.kt:117) ~[bulk-cdk-core-load-0.1.54.jar:?]
	at io.airbyte.cdk.load.task.DestinationTaskLauncher$WrappedTask.execute(DestinationTaskLauncher.kt:124) ~[bulk-cdk-core-load-0.1.54.jar:?]
	at io.airbyte.cdk.load.task.TaskScopeProvider$launch$job$1.invokeSuspend(TaskScopeProvider.kt:35) ~[bulk-cdk-core-load-0.1.54.jar:?]
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33) ~[kotlin-stdlib-2.1.10.jar:2.1.10-release-473]
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:124) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:89) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:586) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:820) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704) ~[kotlinx-coroutines-core-jvm-1.10.2.jar:?]

Stack Trace: 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: [dttp_providers, nationalities, bulk_update_trainee_uploads, degrees, funding_method_subjects, hesa_metadata, funding_methods, nationalisations, apply_applications, courses, withdrawal_reasons, allocation_subjects, bulk_update_row_errors, users, schools, training_partners, subject_specialisms, trainees, placements, academic_cycles, trainee_withdrawal_reasons, trs_trn_requests, course_subjects, potential_duplicate_trainees, dttp_schools, subjects, disabilities, trainee_withdrawals, activities, provider_users, sign_offs, training_partner_users, providers]
	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.task.internal.InputConsumerTask.execute(InputConsumerTask.kt:117)
	at io.airbyte.cdk.load.task.DestinationTaskLauncher$WrappedTask.execute(DestinationTaskLauncher.kt:124)
	at io.airbyte.cdk.load.task.TaskScopeProvider$launch$job$1.invokeSuspend(TaskScopeProvider.kt:35)
	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-02-12 07:06:22 destination INFO main i.a.c.AirbyteConnectorRunnable(run):47 Flushing output consumer prior to shutdown.
2026-02-12 07:06:22 destination INFO main i.a.c.AirbyteConnectorRunnable(run):49 Completed integration: airbyte/destination-bigquery.
2026-02-12 07:06:22 destination ERROR 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: [dttp_providers, nationalities, bulk_update_trainee_uploads, degrees, funding_method_subjects, hesa_metadata, funding_methods, nationalisations, apply_applications, courses, withdrawal_reasons, allocation_subjects, bulk_update_row_errors, users, schools, training_partners, subject_specialisms, trainees, placements, academic_cycles, trainee_withdrawal_reasons, trs_trn_requests, course_subjects, potential_duplicate_trainees, dttp_schools, subjects, disabilities, trainee_withdrawals, activities, provider_users, sign_offs, training_partner_users, providers]
2026-02-12 07:06:22 destination ERROR 	at io.airbyte.cdk.load.state.SyncManager.markInputConsumed(SyncManager.kt:116)
2026-02-12 07:06:22 destination ERROR 	at io.airbyte.cdk.load.state.PipelineEventBookkeepingRouter.close(PipelineEventBookkeepingRouter.kt:300)
2026-02-12 07:06:22 destination ERROR 	at io.airbyte.cdk.load.task.internal.InputConsumerTask.execute(InputConsumerTask.kt:117)
2026-02-12 07:06:22 destination ERROR 	at io.airbyte.cdk.load.task.DestinationTaskLauncher$WrappedTask.execute(DestinationTaskLauncher.kt:124)
2026-02-12 07:06:22 destination ERROR 	at io.airbyte.cdk.load.task.TaskScopeProvider$launch$job$1.invokeSuspend(TaskScopeProvider.kt:35)
2026-02-12 07:06:22 destination ERROR 	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:124)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:89)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:586)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:820)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717)
2026-02-12 07:06:22 destination ERROR 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704)
2026-02-12 07:06:22 replication-orchestrator INFO Destination finished successfully — exiting read dest...
2026-02-12 07:06:22 replication-orchestrator INFO readFromDestination: exception caught
2026-02-12 07:06:22 replication-orchestrator INFO readFromDestination: done. (writeToDestFailed:false, dest.isFinished:true)
2026-02-12 07:06:22 replication-orchestrator INFO Closing StateCheckSumCountEventHandler
2026-02-12 07:06:23 replication-orchestrator INFO sync summary: {
  "status" : "failed",
  "startTime" : 1770878772130,
  "endTime" : 1770879982996,
  "totalStats" : {
    "bytesEmitted" : 0,
    "destinationStateMessagesEmitted" : 0,
    "destinationWriteEndTime" : 0,
    "destinationWriteStartTime" : 1770878772496,
    "meanSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBeforeSourceStateMessageEmitted" : 0,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 0,
    "recordsEmitted" : 0,
    "recordsFilteredOut" : 0,
    "bytesFilteredOut" : 0,
    "replicationEndTime" : 1770879982882,
    "replicationStartTime" : 1770878772130,
    "sourceReadEndTime" : 0,
    "sourceReadStartTime" : 1770878772503,
    "sourceStateMessagesEmitted" : 0
  },
  "streamStats" : [ ],
  "performanceMetrics" : {
    "processFromSource" : {
      "elapsedTimeInNanos" : 514098125,
      "executionCount" : 36,
      "avgExecTimeInNanos" : 1.4280503472222222E7
    },
    "readFromSource" : {
      "elapsedTimeInNanos" : 1203274714739,
      "executionCount" : 219,
      "avgExecTimeInNanos" : 5.494405090132421E9
    },
    "processFromDest" : {
      "elapsedTimeInNanos" : 378926,
      "executionCount" : 1,
      "avgExecTimeInNanos" : 378926.0
    },
    "writeToDest" : {
      "elapsedTimeInNanos" : 0,
      "executionCount" : 0,
      "avgExecTimeInNanos" : "NaN"
    },
    "readFromDest" : {
      "elapsedTimeInNanos" : 1210286266364,
      "executionCount" : 463,
      "avgExecTimeInNanos" : 2.6140092146090713E9
    }
  }
}

A message from kapa.ai