[CI] Failures on 6.x for 60_ml_config_migration/Test old cluster jobs and datafeeds and delete them

[tvernum]

This has failed at least twice since #36698 was merged to 6.x

https://elasticsearch-ci.elastic.co/job/elastic+elasticsearch+6.x+intake/693/console
https://elasticsearch-ci.elastic.co/job/elastic+elasticsearch+6.x+matrix-java-periodic/ES_BUILD_JAVA=java11,ES_RUNTIME_JAVA=java11,nodes=virtual&&linux/127/console

./gradlew :x-pack:qa:rolling-upgrade:with-system-key:v5.6.15#upgradedClusterTestRunner -Dtests.seed=7721FEBE7CCB10B7 -Dtests.class=org.elasticsearch.upgrades.UpgradeClusterClientYamlTestSuiteIT -Dtests.method="test {p0=upgraded_cluster/60_ml_config_migration/Test old cluster jobs and datafeeds and delete them}" -Dtests.security.manager=true -Dtests.locale=ar-AE -Dtests.timezone=Etc/GMT-10 -Dcompiler.java=11 -Druntime.java=8 -Dtests.rest.suite=upgraded_cluster -Dtests.rest.blacklist="/30_ml_jobs_crud/Test model memory limit is updated"

00:50:45    > Throwable #1: java.lang.AssertionError: Failure at [upgraded_cluster/60_ml_config_migration:43]: datafeeds.0.state didn't match expected value:
00:50:45    >              datafeeds.0.state: expected [started] but was [stopped]
00:50:45    > 	at __randomizedtesting.SeedInfo.seed([7721FEBE7CCB10B7:FF75C164D2377D4F]:0)
00:50:45    > 	at org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase.executeSection(ESClientYamlSuiteTestCase.java:406)
00:50:45    > 	at org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase.test(ESClientYamlSuiteTestCase.java:383)
00:50:45    > 	at java.lang.Thread.run(Thread.java:748)
00:50:45    > Caused by: java.lang.AssertionError: datafeeds.0.state didn't match expected value:
00:50:45    >              datafeeds.0.state: expected [started] but was [stopped]
00:50:45    > 	at org.elasticsearch.test.rest.yaml.section.MatchAssertion.doAssert(MatchAssertion.java:87)
00:50:45    > 	at org.elasticsearch.test.rest.yaml.section.Assertion.execute(Assertion.java:76)
00:50:45    > 	at org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase.executeSection(ESClientYamlSuiteTestCase.java:399)
00:50:45    > 	... 38 more

I get failures locally as well but they have a different assertion failure.

I'm going to mute this on 6.x

[elasticmachine]

Pinging @elastic/ml-core

[tvernum]

@davidkyle I assigned this to you since it seems to have come from #36698.
Please reassign if I got that wrong.

[droberts195]

I think this is a case of an old bug, but made far more likely to happen due to the timing changes of having to do a search for job configuration rather than getting it from the cluster state.

This goes back to the age-old inconsistency that when jobs open for the first time they're in the opening state until they are ready to accept data. But when relocating to a new node jobs stay in the opened state, yet cannot accept data until the corresponding C++ process is running on the new node.

We have a check that datafeed persistent tasks cannot be assigned to a node until the associated job is assigned to a node. We also have a check that an assigned datafeed cannot be run until the associated job is in the opened state. If a datafeed's associated job is in the opening state then it waits for a subsequent cluster state update where the job state is opened, but this check assumes that relocating jobs would have a state of opening, which they do not.

Two possible solutions are:

1. Set the job state to opening when it is opened but the assignment is stale
2. Change the test in DatafeedManager to check whether a C++ process is running for the associated job, and not rely on an opened state implying this

[droberts195]

The piece of code I referred to in DatafeedManager is

elasticsearch/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java

Lines 488 to 491 in 0000e7c

	if (jobState == JobState.OPENED) {
	runTask(datafeedTask);
	} else if (jobState == JobState.OPENING) {
	remainingTasks.add(datafeedTask);

[droberts195]

Although option 1 (setting the job state to opening when it is opened but the assignment is stale) seems logically nicer, there is no way to achieve this without making a change to the persistent tasks framework. There is no way to change a task's status on reassignment. Task status can only be set by a separate action, and using that mechanism would leave a period when the same problem that exists now would still exist (just much harder to debug since it would exist for a shorter and less predictable amount of time).