[ML] TransformIT » testStopWaitForCheckpoint test fails

[benwtrent]

Build scan:
https://gradle-enterprise.elastic.co/s/btdxhp7hfpb6y
Repro line:

./gradlew ':x-pack:plugin:transform:qa:multi-node-tests:javaRestTest' --tests "org.elasticsearch.xpack.transform.integration.TransformIT.testStopWaitForCheckpoint" -Dtests.seed=E5346188868E92A8 -Dtests.security.manager=true -Dtests.locale=pt-BR -Dtests.timezone=America/Thule -Druntime.java=8

Reproduces locally?:
No
Applicable branches:
7.x (though probably master as well)
Failure history:

Failure excerpt:

org.elasticsearch.xpack.transform.integration.TransformIT > testStopWaitForCheckpoint FAILED |  
-- | --
  | ElasticsearchStatusException[Elasticsearch exception [type=status_exception, reason=Failed to update transform task [transform-wait-for-checkpoint] state value should_stop_at_checkpoint from [true] to [true]]] |  
  | at __randomizedtesting.SeedInfo.seed([E5346188868E92A8:98E14DE34A1A789E]:0) |  
  | at org.elasticsearch.rest.BytesRestResponse.errorFromXContent(BytesRestResponse.java:187) |  
  | at org.elasticsearch.client.RestHighLevelClient.parseEntity(RestHighLevelClient.java:1911) |  
  | at org.elasticsearch.client.RestHighLevelClient.parseResponseException(RestHighLevelClient.java:1888) |  
  | at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1645) |  
  | at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:1617) |  
  | at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:1584) |  
  | at org.elasticsearch.client.TransformClient.stopTransform(TransformClient.java:319) |  
  | at org.elasticsearch.xpack.transform.integration.TransformIntegTestCase.stopTransform(TransformIntegTestCase.java:136) |  
  | at org.elasticsearch.xpack.transform.integration.TransformIT.testStopWaitForCheckpoint(TransformIT.java:288)

This indicates some race condition when trying to set the value for should_stop_at_checkpoint. I am not sure if it was retried in the call and then caused the failure or if two callers are making the stop call.

Getting logging to attach to the issue.

[elasticmachine]

Pinging @elastic/ml-core (:ml/Transform)

[benwtrent]

transform.zip
Attached are the logs for the test run

[hendrikmuhs]

The log shows an off by 2, so the doc has been updated twice by another caller:

Caused by: org.elasticsearch.index.engine.VersionConflictEngineException: [data_frame_transform_state_and_stats-transform-wait-for-checkpoint]: version conflict, required seqNo [16], primary term [1]. current document has seqNo [18] and primary term [1]

The "other caller" can only be the stop transform transport although it should only update the doc directly if there is no indexer. Looks like some race condition that makes the transport think that there is no indexer.

The logs are not very helpful otherwise, I suggest we add some debug logging to e.g. log whenever the state doc has changed, including seqNo.

[danhermann]

Occurring on 7.10, too: https://gradle-enterprise.elastic.co/s/ofw3rpyu2kupy

[matriv]

and now on 7.12 as well: https://gradle-enterprise.elastic.co/s/jciztdhabcn64

[hendrikmuhs]

relevant log lines from the last failure:

[2021-02-23T10:36:25,835][DEBUG][o.e.x.c.i.AsyncTwoPhaseIndexer] [javaRestTest-0] Finished indexing for job [transform-wait-for-checkpoint], saving state and shutting down.
[2021-02-23T10:36:25,870][DEBUG][o.e.x.t.t.TransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] Querying for changes: {"size":0,"query":{"bool":{"filter":[{"match_all":{"boost":1.0}},{"range":{"timestamp":{"from":1614076574680,"to":1614076584811,"include_lower":true,"include_upper":false,"format":"epoch_millis","boost":1.0}}}],"adjust_pure_negative":true,"boost":1.0}},"aggregations":{"_transform_change_collector":{"composite":{"size":500,"sources":[{"by-business":{"terms":{"field":"business_id","missing_bucket":false,"order":"asc"}}},{"by-user":{"terms":{"field":"user_id","missing_bucket":false,"order":"asc"}}}]}}}}
[2021-02-23T10:36:25,877][DEBUG][o.e.x.t.t.TransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] finished indexing for transform checkpoint [1].
[2021-02-23T10:36:25,879][INFO ][o.e.x.t.t.TransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] transform has stopped.
[2021-02-23T10:36:25,879][DEBUG][o.e.x.t.t.ClientTransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] updating persistent state of transform to [{"task_state":"stopped","indexer_state":"stopped","checkpoint":2,"progress":{"docs_indexed":25,"docs_processed":25},"should_stop_at_checkpoint":false}].
[2021-02-23T10:36:25,892][DEBUG][o.e.x.t.t.ClientTransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] updating persistent state of transform to [{"task_state":"started","indexer_state":"indexing","checkpoint":1,"progress":{"docs_indexed":0,"docs_processed":0},"should_stop_at_checkpoint":true}].
[2021-02-23T10:36:25,922][DEBUG][o.e.x.t.t.ClientTransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] Updated state document from [{seqNo=4,primaryTerm=1,index=.transform-internal-006}] to [{seqNo=7,primaryTerm=1,index=.transform-internal-006}]
[2021-02-23T10:36:25,922][DEBUG][o.e.x.t.t.TransformTask  ] [javaRestTest-0] [transform-wait-for-checkpoint] shutdown of transform requested
[2021-02-23T10:36:25,924][WARN ][o.e.p.AllocatedPersistentTask] [javaRestTest-0] attempt to complete task [data_frame/transforms[c]] with id [transform-wait-for-checkpoint] in the [COMPLETED] state
[2021-02-23T10:36:25,922][ERROR][o.e.x.t.t.ClientTransformIndexer] [javaRestTest-0] [transform-wait-for-checkpoint] updating stats of transform failed.
java.lang.RuntimeException: Failed to persist transform statistics for transform [transform-wait-for-checkpoint]
	at org.elasticsearch.xpack.transform.persistence.IndexBasedTransformConfigManager.lambda$putOrUpdateTransformStoredDoc$14(IndexBasedTransformConfigManager.java:528) ~[?:?]
	at org.elasticsearch.action.ActionListener$1.onFailure(ActionListener.java:125) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]
	at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:39) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]
	at org.elasticsearch.action.support.TransportAction$1.onFailure(TransportAction.java:92) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]
	at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:39) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]
	at org.elasticsearch.action.ActionListener$3.onFailure(ActionListener.java:172) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]
	at org.elasticsearch.action.bulk.TransportSingleItemBulkWriteAction.lambda$wrapBulkResponse$0(TransportSingleItemBulkWriteAction.java:53) ~[elasticsearch-7.12.0-SNAPSHOT.jar:7.12.0-SNAPSHOT]

We see 2 save state calls at the same time, while the second has older parts than the previous one, which indicates 2 threads racing.

Another observation: It looks like the trigger (scheduler) kicked in. I think this is what happens:

• thread A: finish checkpoint 1
• thread A: switch state from indexing to started
• thread A: call doSaveState
• thread A: index call for state persistence with: "task_state":"stopped","indexer_state":"stopped","checkpoint":2
• thread B: trigger, state switches from started to indexing
• thread B: query, search, call doSaveState
• thread B: index call for state persistence with: "task_state":"started","indexer_state":"started","checkpoint":1
• thread A: index call response, set seq_no/primary_id
• thread B: index call response with version conflict

With other words: while checkpoint 1 finishes, the scheduler triggers another run. The race can happen between switching to stopped and the state actually finished. The indexer has a member called indexerThreadShuttingDown, it seems to me, we need to check this in maybeTriggerAsyncJob.