Correctly retrieve data stream write index in `DataStreamsUpgradeIT` (#133433)

The `testUpgradeDataStream` test configures an ILM policy with force
merge on a data stream, which is then reindexed after a cluster upgrade.
The test used to determine the write index of the data stream by looking
at the backing index of the data stream in the cluster state metadata
and retrieving the index that was created last. While this may have
worked for this test suite thusfar, this is generally not the right way
to determine the write index of a data stream, as newer indices can be
put inside the data stream either manually or by ILM.

The latter is the case with the upcoming force merge improvements, where
we perform force merge on a cloned index, meaning the cloned index might
be newer than the write index, resulting in a test failure because the
real write index is not skipped in the `checkILMPhase` method and thus
fails on the ILM `phase` assertion.

We also improve the assertion message to include some more information
to aid test failure investigation.

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

@@ -208,19 +208,21 @@ public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
             createDataStreamFromNonDataStreamIndices(dataStreamFromNonDataStreamIndices);
         } else if (CLUSTER_TYPE == ClusterType.UPGRADED) {
             Map<String, Map<String, Object>> oldIndicesMetadata = getIndicesMetadata(dataStreamName);
+            String oldWriteIndex = getDataStreamBackingIndexNames(dataStreamName).getLast();
             upgradeDataStream(dataStreamName, numRollovers, numRollovers + 1, 0, ilmEnabled);
             cancelReindexTask(dataStreamName);
             upgradeDataStream(dataStreamFromNonDataStreamIndices, 0, 1, 0, ilmEnabled);
             cancelReindexTask(dataStreamFromNonDataStreamIndices);
             Map<String, Map<String, Object>> upgradedIndicesMetadata = getIndicesMetadata(dataStreamName);
+            String newWriteIndex = getDataStreamBackingIndexNames(dataStreamName).getLast();
 
             if (ilmEnabled) {
-                checkILMPhase(dataStreamName, upgradedIndicesMetadata);
+                checkILMPhase(dataStreamName, newWriteIndex);
                 // Delete the data streams to avoid ILM continuously running cluster state tasks, see
                 // https://github.com/elastic/elasticsearch/issues/129097#issuecomment-3016122739
                 deleteDataStream(dataStreamName);
             } else {
-                compareIndexMetadata(oldIndicesMetadata, upgradedIndicesMetadata);
+                compareIndexMetadata(oldIndicesMetadata, oldWriteIndex, upgradedIndicesMetadata);
             }
         }
     }
@@ -262,9 +264,9 @@ public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
 
     private void compareIndexMetadata(
         Map<String, Map<String, Object>> oldIndicesMetadata,
+        String oldWriteIndex,
         Map<String, Map<String, Object>> upgradedIndicesMetadata
     ) {
-        String oldWriteIndex = getWriteIndexFromDataStreamIndexMetadata(oldIndicesMetadata);
         for (Map.Entry<String, Map<String, Object>> upgradedIndexEntry : upgradedIndicesMetadata.entrySet()) {
             String upgradedIndexName = upgradedIndexEntry.getKey();
             if (upgradedIndexName.startsWith(".migrated-")) {
@@ -287,10 +289,8 @@ public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
     }
 
     @SuppressWarnings("unchecked")
-    private void checkILMPhase(String dataStreamName, Map<String, Map<String, Object>> upgradedIndicesMetadata) throws Exception {
-        var writeIndex = getWriteIndexFromDataStreamIndexMetadata(upgradedIndicesMetadata);
+    private void checkILMPhase(String dataStreamName, String writeIndex) throws Exception {
         assertBusy(() -> {
-
             Request request = new Request("GET", dataStreamName + "/_ilm/explain");
             Response response = client().performRequest(request);
             Map<String, Object> responseMap = XContentHelper.convertToMap(
@@ -302,21 +302,12 @@ public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
             for (var index : indices.keySet()) {
                 if (index.equals(writeIndex) == false) {
                     Map<String, Object> ilmInfo = (Map<String, Object>) indices.get(index);
-                    assertThat("Index has not moved to cold ILM phase", ilmInfo.get("phase"), equalTo("cold"));
+                    assertThat("Index [" + index + "] has not moved to cold ILM phase, " + indices, ilmInfo.get("phase"), equalTo("cold"));
                 }
             }
         }, 30, TimeUnit.SECONDS);
     }
 
-    private String getWriteIndexFromDataStreamIndexMetadata(Map<String, Map<String, Object>> indexMetadataForDataStream) {
-        return indexMetadataForDataStream.entrySet()
-            .stream()
-            .sorted((o1, o2) -> Long.compare(getCreationDate(o2.getValue()), getCreationDate(o1.getValue())))
-            .map(Map.Entry::getKey)
-            .findFirst()
-            .get();
-    }
-
     private void startILM() throws IOException {
         setILMInterval();
         var request = new Request("POST", "/_ilm/start");