[destination-mongodb] Out of memory error when copying from temporary to permanent collection
- Linguagem predominante
- Python
- Estrelas
- 22.1k
- Forks
- 5.3k
- Métricas de merge de PRs
- Métricas de PR pendentes
Descrição
### Connector Name
destination-mongodb
### Connector Version
0.2.0
### What step the error happened?
During the sync
### Relevant information
Hello,
I am reporting this out of memory error which happens with the MongoDB destination connector.
The error happens at the end of a synchronisation job, when the mongoDB destination connector tries to copy all the synchronised records from a temporary collection to a permanent.
I believe this error is caused by the copyTable function of the mongodb connector: [link to source code](https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/destination-mongodb/src/main/java/io/airbyte/integrations/destination/mongodb/MongodbRecordConsumer.java#L146C1-L159C4)
All the documents to copy are put in a list in memory to be inserted in the new collection, without any batching. So if there are too many documents, the local memory can be overfilled by this huge list.
This is consistent with my experience where the error happens when I am trying to synchronise a high volume data.
I suggest to fix this by adding some batching. The following code snippet would use a batch size which is fixed size in terms of number of documents:
```java
private static void copyTable(final MongoDatabase mongoDatabase, final String collectionName, final String tmpCollectionName) {
final var tempCollection = mongoDatabase.getOrCreateNewCollection(tmpCollectionName);
final var collection = mongoDatabase.getOrCreateNewCollection(collectionName);
// Define batch size
final int BATCH_SIZE = 1000;
// Create a temporary list to hold the current batch of documents
final List batch = new ArrayList<>();
try (final MongoCursor cursor = tempCollection.find().projection(excludeId()).iterator()) {
while (cursor.hasNext()) {
batch.add(cursor.next());
// When the batch size is reached, insert the batch and clear the list
if (batch.size() == BATCH_SIZE) {
collection.insertMany(new ArrayList<>(batch));
batch.clear();
}
}
// Insert remaining documents that didn't fill a complete batch
if (!batch.isEmpty()) {
collection.insertMany(batch);
}
}
}
```
The code snippet I give would provide some improvements over the current behavior.
The best would be to have a batch which has a fixed size of memory, and not just a fixed size of number of documents (as the size of each mongo document may vary).
### Relevant log output
```shell
2024-12-05 20:00:13 replication-orchestrator INFO Stream Status Update Received: orders - COMPLETE
2024-12-05 20:00:13 replication-orchestrator INFO Updating status: orders - COMPLETE
2024-12-05 20:00:17 destination INFO i.a.i.b.FailureTrackingAirbyteMessageConsumer(close):80 Airbyte message consumer: succeeded.
2024-12-05 20:00:17 destination INFO i.a.i.d.m.MongodbRecordConsumer(close):90 Migration finished with no explicit errors. Copying data from tmp tables to permanent
2024-12-05 20:09:11 destination INFO Malformed non-Airbyte record (connectionId = 9bb79ad9-f258-4d97-bbe6-03f691541e9e): Terminating due to java.lang.OutOfMemoryError: Java heap space
2024-12-05 20:09:12 replication-orchestrator INFO Destination finished successfully — exiting read dest...
2024-12-05 20:09:12 replication-orchestrator INFO readFromDestination: exception caught
2024-12-05 20:09:12 replication-orchestrator INFO readFromDestination: done. (writeToDestFailed:false, dest.isFinished:true)
2024-12-05 20:09:12 replication-orchestrator INFO Closing StateCheckSumCountEventHandler
```
### Contribute
- [x] Yes, I want to contribute
Guia de contribuição
Avaliação
Esta issue ainda não foi avaliada.