[
https://issues.apache.org/jira/browse/FLINK-39398?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski reassigned FLINK-39398:
--------------------------------------
Assignee: Aleksandr Savonin
> Handle duplicate key errors in MongoWriter bulk write retry loop
> ----------------------------------------------------------------
>
> Key: FLINK-39398
> URL: https://issues.apache.org/jira/browse/FLINK-39398
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / MongoDB
> Reporter: Aleksandr Savonin
> Assignee: Aleksandr Savonin
> Priority: Major
>
> MongoWriter.doBulkWrite() catches MongoException broadly without inspecting
> MongoBulkWriteException.getWriteErrors(). This causes two problems when bulk
> writes encounter duplicate key errors (E11000):
> *Problem 1: Infinite restart loop*
> When using InsertOneModel with a unique index and AT_LEAST_ONCE delivery
> guarantee, an unplanned failure (TaskManager kill, OOM, crash) causes Flink
> to replay records from the last checkpoint. Some of these records may already
> exist in MongoDB because they were written before the crash but the source
> offset was not yet committed. The retry loop retries the identical batch
> unchanged, all retries fail with the same E11000 errors, and after exhausting
> maxRetries throws IOException. Flink restarts from checkpoint, replays the
> same records, and the cycle repeats indefinitely.
> *Problem 2: Un-attempted operations dropped with ordered=true*
> With ordered=true (the default), MongoDB stops at the first error. Operations
> after the error index are never attempted by the server. The current code
> retries the entire batch unchanged without re-queuing un-attempted
> operations, which can lead to data loss.
>
> *Solution*
> Add a configurable sink.duplicate-key-handling option with two strategies:
> - fail (default): *Preserves* existing retry behavior. Improves the error
> message to suggest skip-duplicates or upsert mode when E11000 is detected.
> - skip-duplicates: Inspects MongoBulkWriteException.getWriteErrors()
> per-operation. Skips duplicate key errors. For ordered=true, re-queues
> un-attempted operations after the error index. For ordered=false, considers
> non-duplicate operations as already succeeded (no re-queuing).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
