Import is blocked after some user task records are lost

[sdorokhova]

Describe the bug

In some exceptional cases it may happen that part of the data will be lost. We need to make sure that import is able to proceed without being blocked even after data loss. In case of user tasks, if we missed user task CREATED record, the import will be blocked, because the following records for the same user task will try to update non-existent record. We need to use upsert query instead of update.

To Reproduce

Steps to reproduce the behavior:

1. Create user tasks in Zeebe
2. Delete some of the user task creation Zeebe records
3. Start Operate

Current behavior

Import is blocked with exception:

org.elasticsearch.ElasticsearchException: Elasticsearch exception [type=document_missing_exception, reason=[2251799817173350]: document missing]
	at org.elasticsearch.ElasticsearchException.innerFromXContent(ElasticsearchException.java:520) ~[elasticsearch-7.17.19.jar!/:7.17.19]
	at org.elasticsearch.ElasticsearchException.fromXContent(ElasticsearchException.java:431) ~[elasticsearch-7.17.19.jar!/:7.17.19]
	at org.elasticsearch.action.bulk.BulkItemResponse.fromXContent(BulkItemResponse.java:128) ~[elasticsearch-7.17.19.jar!/:7.17.19]
	at org.elasticsearch.action.bulk.BulkResponse.fromXContent(BulkResponse.java:184) ~[elasticsearch-7.17.19.jar!/:7.17.19]
	at org.elasticsearch.client.RestHighLevelClient.parseEntity(RestHighLevelClient.java:2484) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at org.elasticsearch.client.RestHighLevelClient.lambda$performRequestAndParseEntity$8(RestHighLevelClient.java:2105) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:2188) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:2137) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:2105) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at org.elasticsearch.client.RestHighLevelClient.bulk(RestHighLevelClient.java:620) ~[elasticsearch-rest-high-level-client-7.17.19.jar!/:8.9.2]
	at io.camunda.operate.util.ElasticsearchUtil.processLimitedBulkRequest(ElasticsearchUtil.java:399) ~[operate-schema-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.util.ElasticsearchUtil.processBulkRequest(ElasticsearchUtil.java:337) ~[operate-schema-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.util.ElasticsearchUtil.processBulkRequest(ElasticsearchUtil.java:322) ~[operate-schema-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.store.elasticsearch.ElasticsearchBatchRequest.execute(ElasticsearchBatchRequest.java:227) ~[operate-schema-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.AbstractImportBatchProcessor.lambda$performImport$1(AbstractImportBatchProcessor.java:48) ~[operate-importer-common-8.5.0.jar!/:8.5.0]
	at io.micrometer.core.instrument.AbstractTimer.recordCallable(AbstractTimer.java:175) ~[micrometer-core-1.12.4.jar!/:1.12.4]
	at io.camunda.operate.zeebeimport.AbstractImportBatchProcessor.withTimer(AbstractImportBatchProcessor.java:78) ~[operate-importer-common-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.AbstractImportBatchProcessor.withImportIndexQueryTimer(AbstractImportBatchProcessor.java:65) ~[operate-importer-common-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.AbstractImportBatchProcessor.performImport(AbstractImportBatchProcessor.java:46) ~[operate-importer-common-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.ImportJob.processOneIndexBatch(ImportJob.java:161) ~[operate-importer-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.ImportJob.call(ImportJob.java:80) ~[operate-importer-8.5.0.jar!/:8.5.0]
	at io.camunda.operate.zeebeimport.elasticsearch.ElasticsearchRecordsReader.lambda$executeJob$5(ElasticsearchRecordsReader.java:561) ~[operate-importer-8.5.0.jar!/:8.5.0]
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	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:840) [?:?]

Expected behavior

Import can proceed even in case of data loss.

Environment

• Operate Version: 8.5.0

Additional context

• Related incident: https://camunda.slack.com/archives/C06U2APQFRV

---

Acceptance Criteria

Definition of Ready - Checklist

• The bug has been reproduced by the assignee in the environment compatible with the provided one; otherwise, the issue is closed with a comment
• The issue has a meaningful title, description, and testable acceptance criteria
• The issue has been labeled with an appropriate Bug-area label
• Necessary screenshots, screen recordings, or files are attached to the bug report

For UI changes required to solve the bug:

• Design input has been collected by the assignee

Implementation

🔍 Root Cause Analysis

💭 Proposed Solution

👉 Handover Dev to QA

• Changed components:
• Side effects on other components:
• Handy resources:
  BPMN/DMN models, plugins, scripts, REST API endpoints + example payload, etc :
• Example projects:

• Commands/Steps needed to test; Versions to validate:

• Docker file / HELM chart : in case that it needed to be tested via docker share the version contain the fixed along with version of other services .

• Release version ( in which version this fixed/feature will be released):

📗 Link to the test case