[ML] fixing snapshot upgrader bwc test (#65837)

When upgrading a snapshot, the flush request was ignored because headers were not written first.

This commit ensures that the autodetect process headers are written before the flush request is made.

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/JobModelSnapshotUpgrader.java

@@ -233,9 +233,10 @@ public final class JobModelSnapshotUpgrader {
                 return;
             }
             submitOperation(() -> {
+                writeHeader();
                 String flushId = process.flushJob(FlushJobParams.builder().waitForNormalization(false).build());
                 return waitFlushToCompletion(flushId);
-            }, (aVoid, e) -> {
+            }, (flushAcknowledgement, e) -> {
                 Runnable nextStep;
                 if (e != null) {
                     logger.error(
@@ -250,6 +251,12 @@ public final class JobModelSnapshotUpgrader {
                         ActionListener.wrap(t -> shutdown(e), f -> shutdown(e))
                     );
                 } else {
+                    logger.debug(() -> new ParameterizedMessage(
+                        "[{}] [{}] flush [{}] acknowledged requesting state write",
+                        jobId,
+                        snapshotId,
+                        flushAcknowledgement.getId()
+                    ));
                     nextStep = this::requestStateWrite;
                 }
                 threadPool.executor(UTILITY_THREAD_POOL_NAME).execute(nextStep);
@@ -267,7 +274,6 @@ public final class JobModelSnapshotUpgrader {
                         }
                         submitOperation(
                             () -> {
-                                writeHeader();
                                 process.persistState(
                                     params.modelSnapshot().getTimestamp().getTime(),
                                     params.modelSnapshot().getSnapshotId(),

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/MlJobSnapshotUpgradeIT.java

@@ -87,7 +87,6 @@ public class MlJobSnapshotUpgradeIT extends AbstractUpgradeTestCase {
      * index mappings when it is assigned to an upgraded node even if no other ML endpoint is called after the upgrade
      */
     public void testSnapshotUpgrader() throws Exception {
-        assumeTrue("Muting test as failing after https://github.com/elastic/elasticsearch/pull/65755", false);
         hlrc = new HLRC(client()).machineLearning();
         Request adjustLoggingLevels = new Request("PUT", "/_cluster/settings");
         adjustLoggingLevels.setJsonEntity(
@@ -101,6 +100,11 @@ public class MlJobSnapshotUpgradeIT extends AbstractUpgradeTestCase {
                 break;
             case MIXED:
                 assumeTrue("We should only test if old cluster is before new cluster", UPGRADE_FROM_VERSION.before(Version.CURRENT));
+                ensureHealth((request -> {
+                    request.addParameter("timeout", "70s");
+                    request.addParameter("wait_for_nodes", "3");
+                    request.addParameter("wait_for_status", "yellow");
+                }));
                 testSnapshotUpgradeFailsOnMixedCluster();
                 break;
             case UPGRADED: