Firestore connector not letting me sync properly

### Connector Name

destination-firebase

### Connector Version

0.2.0

### What step the error happened?

During the sync

### Relevant information

when syncing my airtable data to firestore database it shows an error and fails.

it does seem to be making some data in firestore but it is not syncing as it keeps adding the same data so i end up with multiple copies of one thing

i assume it is something to do with the firestore connector but not 100% sure that is the issue

### Relevant log output

```shell
2025-11-26 17:27:07 destination INFO Begin writing to the destination...
2025-11-26 17:27:07 destination ERROR '_UnaryStreamMultiCallable' object has no attribute '_retry'
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/google/api_core/grpc_helpers.py", line 116, in __next__
return next(self._wrapped)
^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/grpc/_channel.py", line 543, in __next__
return self._next()
^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/grpc/_channel.py", line 952, in _next
raise self
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.DEADLINE_EXCEEDED
details = "Deadline Exceeded"
debug_error_string = "UNKNOWN:Error received from peer {grpc_message:"Deadline Exceeded", grpc_status:4, created_time:"2025-11-26T17:27:07.225064265+00:00"}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py", line 411, in _make_stream
response = next(response_iterator, None)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/google/api_core/grpc_helpers.py", line 119, in __next__
raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/airbyte/integration_code/main.py", line 12, in
DestinationFirestore().run(sys.argv[1:])
File "/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py", line 153, in run
for message in output_messages:
File "/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py", line 143, in run_cmd
yield from self._run_write(
File "/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py", line 66, in _run_write
yield from self.write(
File "/airbyte/integration_code/destination_firestore/destination.py", line 38, in write
writer.purge(configured_stream.stream.name)
File "/airbyte/integration_code/destination_firestore/writer.py", line 35, in purge
for doc in self.client.collection(stream).stream():
File "/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/stream_generator.py", line 58, in __next__
return self._generator.__next__()
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py", line 413, in _make_stream
if self._retry_query_after_exception(exc, retry, transaction):
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py", line 264, in _retry_query_after_exception
retry = gapic_callable._retry
^^^^^^^^^^^^^^^^^^^^^
AttributeError: '_UnaryStreamMultiCallable' object has no attribute '_retry'
2025-11-26 17:27:08 replication-orchestrator ERROR DestinationWriter error:
2025-11-26 17:27:08 replication-orchestrator ERROR DestinationWriter error:
2025-11-26 17:27:08 replication-orchestrator WARN Attempted to close a destination which is already closed.
2025-11-26 17:27:08 replication-orchestrator INFO DestinationWriter finished.
2025-11-26 17:27:08 replication-orchestrator ERROR DestinationReader error:
2025-11-26 17:27:08 replication-orchestrator INFO DestinationReader finished.
2025-11-26 17:27:08 replication-orchestrator ERROR SourceReader error:
2025-11-26 17:27:08 replication-orchestrator INFO SourceReader finished.
2025-11-26 17:27:08 replication-orchestrator INFO MessageProcessor finished.
2025-11-26 17:27:08 replication-orchestrator ERROR runJobs failed; recording failure but continuing to finish.
2025-11-26 17:27:08 replication-orchestrator INFO Closing StateCheckSumCountEventHandler
2025-11-26 17:27:08 replication-orchestrator INFO Sync summary: {
"status" : "failed",
"recordsSynced" : 0,
"bytesSynced" : 0,
"startTime" : 1764177720634,
"endTime" : 1764178028333,
"totalStats" : {
"additionalStats" : { },
"bytesCommitted" : 0,
"bytesEmitted" : 1075009,
"destinationStateMessagesEmitted" : 0,
"destinationWriteEndTime" : 0,
"destinationWriteStartTime" : 1764177727643,
"meanSecondsBeforeSourceStateMessageEmitted" : 0,
"maxSecondsBeforeSourceStateMessageEmitted" : 0,
"maxSecondsBetweenStateMessageEmittedandCommitted" : 0,
"meanSecondsBetweenStateMessageEmittedandCommitted" : 0,
"recordsEmitted" : 1093,
"recordsCommitted" : 0,
"recordsFilteredOut" : 0,
"bytesFilteredOut" : 0,
"replicationEndTime" : 1764178028329,
"replicationStartTime" : 1764177720634,
"sourceReadEndTime" : 0,
"sourceReadStartTime" : 1764177727643,
"sourceStateMessagesEmitted" : 0
},
"streamStats" : [ {
"streamName" : "sync_to_firebase/line_items_sync_view_to_sync_base/tblJ5NVOUQOwFxpPF",
"stats" : {
"additionalStats" : {
"sourceFieldsPopulated" : 13077
},
"bytesCommitted" : 0,
"bytesEmitted" : 625523,
"estimatedBytes" : 0,
"estimatedRecords" : 0,
"recordsEmitted" : 979,
"recordsCommitted" : 0,
"recordsFilteredOut" : 0,
"bytesFilteredOut" : 0
}
}, {
"streamName" : "sync_to_firebase/project_sync_view_to_sync_base/tblrOXILFCWy8gJQo",
"stats" : {
"additionalStats" : {
"sourceFieldsPopulated" : 982
},
"bytesCommitted" : 0,
"bytesEmitted" : 449486,
"estimatedBytes" : 0,
"estimatedRecords" : 0,
"recordsEmitted" : 114,
"recordsCommitted" : 0,
"recordsFilteredOut" : 0,
"bytesFilteredOut" : 0
}
} ],
"performanceMetrics" : { },
"streamCount" : 2
}
2025-11-26 17:27:08 replication-orchestrator INFO Failures: [ {
"failureOrigin" : "destination",
"failureType" : "system_error",
"internalMessage" : "'_UnaryStreamMultiCallable' object has no attribute '_retry'",
"externalMessage" : "Something went wrong in the connector. See the logs for more details.",
"metadata" : {
"attemptNumber" : 4,
"jobId" : 59166295,
"from_trace_message" : true,
"connector_command" : "write"
},
"stacktrace" : "Traceback (most recent call last):\n File \"/usr/local/lib/python3.11/site-packages/google/api_core/grpc_helpers.py\", line 116, in __next__\n return next(self._wrapped)\n ^^^^^^^^^^^^^^^^^^^\n File \"/usr/local/lib/python3.11/site-packages/grpc/_channel.py\", line 543, in __next__\n return self._next()\n ^^^^^^^^^^^^\n File \"/usr/local/lib/python3.11/site-packages/grpc/_channel.py\", line 952, in _next\n raise self\ngrpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:\n\tstatus = StatusCode.DEADLINE_EXCEEDED\n\tdetails = \"Deadline Exceeded\"\n\tdebug_error_string = \"UNKNOWN:Error received from peer {grpc_message:\"Deadline Exceeded\", grpc_status:4, created_time:\"2025-11-26T17:27:07.225064265+00:00\"}\"\n>\n\nThe above exception was the direct cause of the following exception:\n\nTraceback (most recent call last):\n File \"/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py\", line 411, in _make_stream\n response = next(response_iterator, None)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/usr/local/lib/python3.11/site-packages/google/api_core/grpc_helpers.py\", line 119, in __next__\n raise exceptions.from_grpc_error(exc) from exc\ngoogle.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File \"/airbyte/integration_code/main.py\", line 12, in \n DestinationFirestore().run(sys.argv[1:])\n File \"/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py\", line 153, in run\n for message in output_messages:\n File \"/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py\", line 143, in run_cmd\n yield from self._run_write(\n File \"/usr/local/lib/python3.11/site-packages/airbyte_cdk/destinations/destination.py\", line 66, in _run_write\n yield from self.write(\n File \"/airbyte/integration_code/destination_firestore/destination.py\", line 38, in write\n writer.purge(configured_stream.stream.name)\n File \"/airbyte/integration_code/destination_firestore/writer.py\", line 35, in purge\n for doc in self.client.collection(stream).stream():\n File \"/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/stream_generator.py\", line 58, in __next__\n return self._generator.__next__()\n ^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py\", line 413, in _make_stream\n if self._retry_query_after_exception(exc, retry, transaction):\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/usr/local/lib/python3.11/site-packages/google/cloud/firestore_v1/query.py\", line 264, in _retry_query_after_exception\n retry = gapic_callable._retry\n ^^^^^^^^^^^^^^^^^^^^^\nAttributeError: '_UnaryStreamMultiCallable' object has no attribute '_retry'\n",
"timestamp" : 1764178027312
}, {
"failureOrigin" : "replication",
"internalMessage" : "Broken pipe",
"externalMessage" : "Something went wrong during replication",
"metadata" : {
"attemptNumber" : 4,
"jobId" : 59166295
},
"stacktrace" : "java.io.IOException: Broken pipe\n\tat java.base/sun.nio.ch.UnixFileDispatcherImpl.write0(Native Method)\n\tat java.base/sun.nio.ch.UnixFileDispatcherImpl.write(UnixFileDispatcherImpl.java:65)\n\tat java.base/sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:137)\n\tat java.base/sun.nio.ch.IOUtil.write(IOUtil.java:102)\n\tat java.base/sun.nio.ch.IOUtil.write(IOUtil.java:72)\n\tat java.base/sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:300)\n\tat java.base/sun.nio.ch.ChannelOutputStream.writeFully(ChannelOutputStream.java:68)\n\tat java.base/sun.nio.ch.ChannelOutputStream.write(ChannelOutputStream.java:105)\n\tat java.base/sun.nio.cs.StreamEncoder.writeBytes(StreamEncoder.java:309)\n\tat java.base/sun.nio.cs.StreamEncoder.implWrite(StreamEncoder.java:381)\n\tat java.base/sun.nio.cs.StreamEncoder.implWrite(StreamEncoder.java:357)\n\tat java.base/sun.nio.cs.StreamEncoder.lockedWrite(StreamEncoder.java:158)\n\tat java.base/sun.nio.cs.StreamEncoder.write(StreamEncoder.java:139)\n\tat java.base/java.io.OutputStreamWriter.write(OutputStreamWriter.java:219)\n\tat java.base/java.io.BufferedWriter.implFlushBuffer(BufferedWriter.java:178)\n\tat java.base/java.io.BufferedWriter.flushBuffer(BufferedWriter.java:163)\n\tat java.base/java.io.BufferedWriter.implWrite(BufferedWriter.java:334)\n\tat java.base/java.io.BufferedWriter.write(BufferedWriter.java:313)\n\tat java.base/java.io.Writer.write(Writer.java:278)\n\tat io.airbyte.container.orchestrator.worker.io.AirbyteMessageBufferedWriter.write(AirbyteMessageBufferedWriter.kt:27)\n\tat io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteDestination.acceptWithNoTimeoutMonitor(LocalContainerAirbyteDestination.kt:132)\n\tat io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteDestination.accept(LocalContainerAirbyteDestination.kt:92)\n\tat io.airbyte.container.orchestrator.worker.DestinationWriter.run(ReplicationTask.kt:97)\n\tat io.airbyte.container.orchestrator.worker.DestinationWriter$run$1.invokeSuspend(ReplicationTask.kt)\n\tat kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:34)\n\tat kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)\n\tat io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:141)\n\tat io.micrometer.core.instrument.Timer.lambda$wrap$2(Timer.java:199)\n\tat datadog.trace.bootstrap.instrumentation.java.concurrent.Wrapper.run(Wrapper.java:47)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)\n\tat java.base/java.lang.Thread.run(Thread.java:1583)\n",
"timestamp" : 1764178028247
} ]
2025-11-26 17:27:08 replication-orchestrator INFO
2025-11-26 17:27:08 replication-orchestrator INFO Closing io.micronaut.context.propagation.instrument.execution.ContextPropagatingExecutorService@37b23736
2025-11-26 17:27:08 replication-orchestrator INFO ----- END REPLICATION -----
2025-11-26 17:27:08 replication-orchestrator INFO Closing io.micronaut.context.propagation.instrument.execution.ContextPropagatingExecutorService@75798d55
2025-11-26 17:27:08 replication-orchestrator INFO
2025-11-26 17:27:08 replication-orchestrator INFO Closing io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteDestination@6c037fa4
2025-11-26 17:27:08 replication-orchestrator WARN Attempted to close a destination which is already closed.
2025-11-26 17:27:08 replication-orchestrator ERROR Error closing resource io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteDestination@6c037fa4; recording failure but continuing.
2025-11-26 17:27:08 replication-orchestrator INFO Closing io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteSource@53d3faa9
2025-11-26 17:28:08 replication-orchestrator WARN Failed to wait for exit value file /source/exitCode.txt to be found.
2025-11-26 17:28:08 replication-orchestrator ERROR Error closing resource io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteSource@53d3faa9; recording failure but continuing.
2025-11-26 17:28:08 replication-orchestrator INFO Closing io.airbyte.container.orchestrator.worker.RecordSchemaValidator@4d9aaed2
2025-11-26 17:28:08 replication-orchestrator INFO Closing io.airbyte.container.orchestrator.persistence.SyncPersistenceImpl@40a21fe9
2025-11-26 17:28:10 replication-orchestrator INFO Returning output...
2025-11-26 17:28:11 platform INFO
----- START POST REPLICATION OPERATIONS -----

2025-11-26 17:28:11 platform INFO No post-replication operation(s) to perform.
2025-11-26 17:28:11 platform INFO
----- END POST REPLICATION OPERATIONS -----

2025-11-26 17:28:11 platform INFO Retry State: RetryManager(completeFailureBackoffPolicy=BackoffPolicy(minInterval=PT10S, maxInterval=PT30M, base=3), partialFailureBackoffPolicy=null, successiveCompleteFailureLimit=5, successivePartialFailureLimit=1000, totalCompleteFailureLimit=5, totalPartialFailureLimit=20, successiveCompleteFailures=5, successivePartialFailures=0, totalCompleteFailures=5, totalPartialFailures=0)
Backoff before next attempt: 13 minutes 30 seconds
2025-11-26 17:28:11 platform ERROR Failing job: 59166295, reason: Job failed after too many retries for connection b3416e47-2544-4413-9a46-af7d695e0343
```

### Contribute

- [ ] Yes, I want to contribute