Hi, we are using Airbyte in our kubernetes clusters with a connection with BigQuery source (0.1.8 version I think) and Kafka destination (0.1.8 I think). Airbyte is setup with 0.39.1 version. The jobs are run on large machines (16cpu, 32G) as we previously noted that on large batches, the containers were being killed by the OOM reaper. the source table has ~100M records. We run into these non deterministic errors (each attempt failed with different amount of data read) and it seems that it is not reading all the data but still logs “Source has no more messages, closing connection.“; I can confirm that the source pod is eventually stopped, but that seems to be the cause of failure here? The following logs are from Attempt 3. Initial attempts failed at 2M records read and 90M records read.

2022-10-12 15:17:22 [32mINFO[m i.a.w.g.DefaultReplicationWorker(lambda$getReplicationRunnable$6):334 - Records read: 4587000 (655 MB)
2022-10-12 15:17:31 [32mINFO[m i.a.w.g.DefaultReplicationWorker(lambda$getReplicationRunnable$6):337 - Source has no more messages, closing connection.
2022-10-12 15:17:33 [32mINFO[m i.a.w.p.KubePodProcess(close):713 - (pod: t-107 / source-bigquery-read-4529-2-irywx) - Closed all resources for pod
2022-10-12 15:17:33 [32mINFO[m i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. 
errors: $: null found, object expected
2022-10-12 15:17:33 [1;31mERROR[m i.a.w.i.DefaultAirbyteStreamFactory(lambda$create$1):70 - Validation failed: null
2022-10-12 15:17:34 [32mINFO[m i.a.w.p.KubePodProcess(close):713 - (pod: t-107 / destination-kafka-write-4529-2-wnxxp) - Closed all resources for pod
2022-10-12 15:17:34 [1;31mERROR[m i.a.w.g.DefaultReplicationWorker(run):177 - Sync worker failed.
java.util.concurrent.ExecutionException: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396) ~[?:?]
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) ~[?:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:170) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:62) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:158) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
	Suppressed: io.airbyte.workers.exception.WorkerException: Source process exit with code 1. This warning is normal if the job was cancelled.
		at io.airbyte.workers.internal.DefaultAirbyteSource.close(DefaultAirbyteSource.java:136) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:134) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:62) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
		at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:158) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
		at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:341) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	... 1 more
Caused by: io.airbyte.workers.exception.WorkerException: Source process exit with code 1. This warning is normal if the job was cancelled.
	at io.airbyte.workers.internal.DefaultAirbyteSource.close(DefaultAirbyteSource.java:136) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:339) ~[io.airbyte-airbyte-workers-0.39.1-alpha.jar:?]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	... 1 more
2022-10-12 15:17:34 [32mINFO[m i.a.w.g.DefaultReplicationWorker(run):236 - sync summary: io.airbyte.config.ReplicationAttemptSummary@613c6c00[status=failed,recordsSynced=4587950,bytesSynced=687407575,startTime=1665586308868,endTime=1665587854669,totalStats=io.airbyte.config.SyncStats@4fe132af[recordsEmitted=4587950,bytesEmitted=687407575,stateMessagesEmitted=0,recordsCommitted=0],streamStats=[io.airbyte.config.StreamSyncStats@2251ad00[streamName=fennel_impressions,stats=io.airbyte.config.SyncStats@5ffe54a3[recordsEmitted=4587950,bytesEmitted=687407575,stateMessagesEmitted=<null>,recordsCommitted=<null>]]]]
2022-10-12 15:17:34 [32mINFO[m i.a.w.g.DefaultReplicationWorker(run):265 - Source did not output any state messages
2022-10-12 15:17:34 [33mWARN[m i.a.w.g.DefaultReplicationWorker(run):273 - State capture: No new state, falling back on input state: io.airbyte.config.State@47f56920[state={"cdc":false,"streams":[{"cursor":"0","stream_name":"fennel_impressions","cursor_field":["added_on"],"stream_namespace":"impression_dataset"}]}]
2022-10-12 15:17:34 [32mINFO[m i.a.w.t.TemporalAttemptExecution(get):134 - Stopping cancellation check scheduling...
2022-10-12 15:17:34 [32mINFO[m i.a.w.t.s.ReplicationActivityImpl(lambda$replicate$3):157 - sync summary: io.airbyte.config.StandardSyncOutput@5f71a285[standardSyncSummary=io.airbyte.config.StandardSyncSummary@47bc3f11[status=failed,recordsSynced=4587950,bytesSynced=687407575,startTime=1665586308868,endTime=1665587854669,totalStats=io.airbyte.config.SyncStats@4fe132af[recordsEmitted=4587950,bytesEmitted=687407575,stateMessagesEmitted=0,recordsCommitted=0],streamStats=[io.airbyte.config.StreamSyncStats@2251ad00[streamName=fennel_impressions,stats=io.airbyte.config.SyncStats@5ffe54a3[recordsEmitted=4587950,bytesEmitted=687407575,stateMessagesEmitted=<null>,recordsCommitted=<null>]]]],normalizationSummary=<null>,state=io.airbyte.config.State@47f56920[state={"cdc":false,"streams":[{"cursor":"0","stream_name":"fennel_impressions","cursor_field":["added_on"],"stream_namespace":"impression_dataset"}]}],outputCatalog=io.airbyte.protocol.models.ConfiguredAirbyteCatalog@3d45e59d[streams=[io.airbyte.protocol.models.ConfiguredAirbyteStream@1aaa2f0[stream=io.airbyte.protocol.models.AirbyteStream@539d5dca[name=fennel_impressions,jsonSchema={"type":"object","properties":{"post_id":{"type":"number"},"user_id":{"type":"number"},"added_on":{"type":"number"},"event_name":{"type":"string"},"event_timestamp":{"type":"number"},"time_spent_secs":{"type":"number"}}},supportedSyncModes=[full_refresh, incremental],sourceDefinedCursor=<null>,defaultCursorField=[],sourceDefinedPrimaryKey=[],namespace=impression_dataset,additionalProperties={}],syncMode=incremental,cursorField=[added_on],destinationSyncMode=append,primaryKey=[],additionalProperties={}]],additionalProperties={}],failures=[io.airbyte.config.FailureReason@757fb41f[failureOrigin=source,failureType=<null>,internalMessage=io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!,externalMessage=Something went wrong within the source connector,metadata=io.airbyte.config.Metadata@6704de20[additionalProperties={attemptNumber=2, jobId=4529}],stacktrace=java.util.concurrent.CompletionException: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1807)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:341)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
	... 3 more
Caused by: io.airbyte.workers.exception.WorkerException: Source process exit with code 1. This warning is normal if the job was cancelled.
	at io.airbyte.workers.internal.DefaultAirbyteSource.close(DefaultAirbyteSource.java:136)
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:339)
	... 4 more
,retryable=<null>,timestamp=1665587853062]]]
2022-10-12 15:17:34 [32mINFO[m i.a.w.t.TemporalUtils(withBackgroundHeartbeat):236 - Stopping temporal heartbeating...
2022-10-12 15:17:34 [32mINFO[m i.a.c.p.ConfigRepository(updateConnectionState):774 - Updating connection 1a755743-0802-4dea-a026-29071d0c55db state: io.airbyte.config.State@2e6b37f2[state={"cdc":false,"streams":[{"cursor":"0","stream_name":"fennel_impressions","cursor_field":["added_on"],"stream_namespace":"impression_dataset"}]}]
2022-10-12 15:17:13 [44msource[0m > 2022-10-12 15:17:13 [32mINFO[m i.a.i.s.r.AbstractDbSource(lambda$createReadIterator$7):250 - Reading stream fennel_impressions. Records read: 4530000
...
2022-10-12 15:17:21 [44msource[0m > 2022-10-12 15:17:20 [32mINFO[m i.a.i.s.r.AbstractDbSource(lambda$createReadIterator$7):250 - Reading stream fennel_impressions. Records read: 4580000
2022-10-12 15:17:31 [44msource[0m > 2022-10-12 15:17:31 [32mINFO[m i.a.i.s.r.AbstractDbSource(lambda$read$2):124 - Closing database connection pool.
2022-10-12 15:17:31 [44msource[0m > 2022-10-12 15:17:31 [32mINFO[m i.a.i.s.r.AbstractDbSource(lambda$read$2):126 - Closed database connection pool.
2022-10-12 15:17:31 [44msource[0m > Exception in thread "main" com.google.cloud.bigquery.BigQueryException: <http://www.googleapis.com|www.googleapis.com>
2022-10-12 15:17:31 [44msource[0m > 	at com.google.cloud.bigquery.spi.v2.HttpBigQueryRpc.translate(HttpBigQueryRpc.java:115)
2022-10-12 15:17:31 [44msource[0m > 	at com.google.cloud.bigquery.spi.v2.HttpBigQueryRpc.listTableData(HttpBigQueryRpc.java:514)
2022-10-12 15:17:31 [44msource[0m > 	at com.google.cloud.bigquery.BigQueryImpl$29.call(BigQueryImpl.java:1129)
2022-10-12 15:17:31 [44msource[0m > 	at com.google.cloud.bigquery.BigQueryImpl$29.call(BigQueryImpl.java:1124)
......
2022-10-12 15:17:31 [44msource[0m > 	... 53 more
2022-10-12 15:17:33 [43mdestination[0m > 2022-10-12 15:17:33 [32mINFO[m i.a.i.b.FailureTrackingAirbyteMessageConsumer(close):65 - Airbyte message consumer: succeeded.
2022-10-12 15:17:33 [43mdestination[0m > 2022-10-12 15:17:33 [32mINFO[m i.a.i.b.IntegrationRunner(runInternal):171 - Completed integration: io.airbyte.integrations.destination.kafka.KafkaDestination
2022-10-12 15:17:33 [43mdestination[0m > 2022-10-12 15:17:33 [32mINFO[m i.a.i.d.k.KafkaDestination(main):86 - Completed destination: class io.airbyte.integrations.destination.kafka.KafkaDestination

@[DEPRECATED] Marcos Marx turned this message into Zendesk ticket 2780 to ensure timely resolution!

Hi. It fails each time on different stage since it chunks the data and reads in parallel (most likely - though I am taking a look at the source code as we speak) This is the error:

errors: $: null found, object expected

did you have a run that it succeeded?

Yeah we did have runs in the past which succeeded with a smaller dataset. I don’t see this error message there. However I do see this -

2022-10-12 06:06:33 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. 
errors: $.security_protocol: does not have a value in the enumeration [PLAINTEXT]
2022-10-12 06:06:33 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. 
errors: $.sasl_mechanism: does not have a value in the enumeration [PLAIN], $.security_protocol: does not have a value in the enumeration [SASL_PLAINTEXT], $.sasl_jaas_config: object found, string expected
2022-10-12 06:06:33 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. 
errors: $.sasl_jaas_config: object found, string expected

potentially coming in from the Kafka configuration, but it succeeds. So not sure if this error log is helpful. The last logs of the last successful run -

2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.s.r.AbstractDbSource(lambda$createReadIterator$7):250 - Reading stream fennel_impressions. Records read: 190000
2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.s.r.StateDecoratingIterator(computeNext):60 - State Report: stream name: AirbyteStreamNameNamespacePair{name='fennel_impressions', namespace='impression_dataset'}, original cursor field: added_on, original cursor 1665552014312838, cursor field: added_on, new cursor: 1665553029829536
2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.s.r.AbstractDbSource(lambda$read$2):124 - Closing database connection pool.
2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.s.r.AbstractDbSource(lambda$read$2):126 - Closed database connection pool.
2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.b.IntegrationRunner(runInternal):169 - Completed integration: io.airbyte.integrations.source.bigquery.BigQuerySource
2022-10-12 06:10:25 source > 2022-10-12 06:10:25 INFO i.a.i.s.b.BigQuerySource(main):176 - completed source: class io.airbyte.integrations.source.bigquery.BigQuerySource
2022-10-12 06:10:26 destination > 2022-10-12 06:10:26 INFO i.a.i.b.FailureTrackingAirbyteMessageConsumer(close):65 - Airbyte message consumer: succeeded.
2022-10-12 06:10:26 destination > 2022-10-12 06:10:26 INFO i.a.i.b.IntegrationRunner(runInternal):171 - Completed integration: io.airbyte.integrations.destination.kafka.KafkaDestination
2022-10-12 06:10:26 destination > 2022-10-12 06:10:26 INFO i.a.i.d.k.KafkaDestination(main):86 - Completed destination: class io.airbyte.integrations.destination.kafka.KafkaDestination

So the are few errors that I can see. 1. 2022-10-12 151733 [32mINFO[m i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. errors: $: null found, object expected - this reflects the other errors 2. 2022-10-12 060633 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. errors: $.security_protocol: does not have a value in the enumeration [PLAINTEXT] 3. 2022-10-12 060633 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. errors: $.sasl_mechanism: does not have a value in the enumeration [PLAIN], $.security_protocol: does not have a value in the enumeration [SASL_PLAINTEXT], $.sasl_jaas_config: object found, string expected 4. 2022-10-12 060633 INFO i.a.v.j.JsonSchemaValidator(test):56 - JSON schema validation failed. errors: $.sasl_jaas_config: object found, string expected all the error are because of a value that is an object and not a string and it does not pass the json schema validation test 2-4 errors are java kafka api errors it fails here:

switch (protocol) {
      case PLAINTEXT -> {}
      case SASL_SSL, SASL_PLAINTEXT -> {
        builder.put(SaslConfigs.SASL_JAAS_CONFIG, protocolConfig.get("sasl_jaas_config").asText());
        builder.put(SaslConfigs.SASL_MECHANISM, protocolConfig.get("sasl_mechanism").asText());
      }
      default -> throw new RuntimeException("Unexpected Kafka protocol: " + Jsons.serialize(protocol));
    }

so the protocol in the json schema must be an object of strings. in this case it`s hard for me to exactly say WHERE is the issue located from your source. one of the protocol fields in null some where i would open an issue here though before that validate that all protocol fields are correct as to why is runs ok some times - hard to say

@Mohit Reddy there's been several recent improvements to BigQuery's source connector, if you are on 0.1.8 can you try a more recent version and see if this helps? Currently at 0.2.2 https://docs.airbyte.com/integrations/sources/bigquery#source-bigquery

one of the protocol fields in null some where

protocol fields? If the configuration fields are incorrect, we would see the failure consistently yeah? Could it be because one of the fields in the data is null? Maybe the field which is configured as the “cursor”?

It is difficult to say where in the data and which one is null - if you can upgrade the source and destination and airbyte overall - give it a go - It could be code issue as well

Post upgrading, we see a consistent behavior. Thanks for the suggestions.

I'm not 100% sure but possibly this one added to BigQuery source 0.2.0 https://github.com/airbytehq/airbyte/pull/14362