Retry synced-flush on conflict in tests (#66968)

Closes #66631

Author: dnhatn

qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java

@@ -731,13 +731,7 @@ public void testRecovery() throws Exception {
             assertOK(client().performRequest(flushRequest));
 
             if (randomBoolean()) {
-                // We had a bug before where we failed to perform peer recovery with sync_id from 5.x to 6.x.
-                // We added this synced flush so we can exercise different paths of recovery code.
-                try {
-                    performSyncedFlush(index);
-                } catch (ResponseException ignored) {
-                    // synced flush is optional here
-                }
+                performSyncedFlush(index, randomBoolean());
             }
             if (shouldHaveTranslog) {
                 // Update a few documents so we are sure to have a translog
@@ -1421,7 +1415,6 @@ public void testTurnOffTranslogRetentionAfterUpgraded() throws Exception {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/66631")
     public void testRecoveryWithTranslogRetentionDisabled() throws Exception {
         if (isRunningAgainstOldCluster()) {
             final Settings.Builder settings = Settings.builder()
@@ -1452,7 +1445,7 @@ public void testRecoveryWithTranslogRetentionDisabled() throws Exception {
             if (randomBoolean()) {
                 flush(index, randomBoolean());
             } else if (randomBoolean()) {
-                performSyncedFlush(index);
+                performSyncedFlush(index, randomBoolean());
             }
             saveInfoDocument("doc_count", Integer.toString(numDocs));
         }

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java

@@ -308,7 +308,8 @@ public void testRelocationWithConcurrentIndexing() throws Exception {
                 throw new IllegalStateException("unknown type " + CLUSTER_TYPE);
         }
         if (randomBoolean()) {
-            syncedFlush(index);
+            performSyncedFlush(index, randomBoolean());
+            ensureGlobalCheckpointSynced(index);
         }
     }
 
@@ -587,22 +588,6 @@ private void assertClosedIndex(final String index, final boolean checkRoutingTab
         }
     }
 
-    private void syncedFlush(String index) throws Exception {
-        // We have to spin synced-flush requests here because we fire the global checkpoint sync for the last write operation.
-        // A synced-flush request considers the global checkpoint sync as an going operation because it acquires a shard permit.
-        assertBusy(() -> {
-            try {
-                Response resp = performSyncedFlush(index);
-                Map<String, Object> result = ObjectPath.createFromResponse(resp).evaluate("_shards");
-                assertThat(result.get("failed"), equalTo(0));
-            } catch (ResponseException ex) {
-                throw new AssertionError(ex); // cause assert busy to retry
-            }
-        });
-        // ensure the global checkpoint is synced; otherwise we might trim the commit with syncId
-        ensureGlobalCheckpointSynced(index);
-    }
-
     @SuppressWarnings("unchecked")
     private void assertPeerRecoveredFiles(String reason, String index, String targetNode, Matcher<Integer> sizeMatcher) throws IOException {
         Map<?, ?> recoveryStats = entityAsMap(client().performRequest(new Request("GET", index + "/_recovery")));
@@ -668,7 +653,8 @@ public void testUpdateDoc() throws Exception {
             assertThat(XContentMapValues.extractValue("_source.updated_field", doc), equalTo(updates.get(docId)));
         }
         if (randomBoolean()) {
-            syncedFlush(index);
+            performSyncedFlush(index, randomBoolean());
+            ensureGlobalCheckpointSynced(index);
         }
     }
 

qa/translog-policy/src/test/java/org/elasticsearch/upgrades/TranslogPolicyIT.java

@@ -132,7 +132,7 @@ public void testRecoverReplica() throws Exception {
             if (randomBoolean()) {
                 flush(index, randomBoolean());
             } else if (randomBoolean()) {
-                performSyncedFlush(index);
+                performSyncedFlush(index, randomBoolean());
             }
         }
         ensureGreen(index);

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -1502,7 +1502,7 @@ protected static Version minimumNodeVersion() throws IOException {
         return minVersion;
     }
 
-    protected static Response performSyncedFlush(String indexName) throws IOException {
+    protected static void performSyncedFlush(String indexName, boolean retryOnConflict) throws Exception {
         final Request request = new Request("POST", indexName + "/_flush/synced");
         final List<String> expectedWarnings = Collections.singletonList(SyncedFlushService.SYNCED_FLUSH_DEPRECATION_MESSAGE);
         // prior to v7.11, the deprecation message for synced flush was incorrect so we also need to handle that
@@ -1519,7 +1519,22 @@ protected static Response performSyncedFlush(String indexName) throws IOExceptio
                 warnings.equals(expectedWarnings) == false && warnings.equals(incorrectWarningMessage) == false);
         }
         request.setOptions(options);
-        return client().performRequest(request);
+        // We have to spin a synced-flush request because we fire the global checkpoint sync for the last write operation.
+        // A synced-flush request considers the global checkpoint sync as an going operation because it acquires a shard permit.
+        assertBusy(() -> {
+            try {
+                Response resp = client().performRequest(request);
+                if (retryOnConflict) {
+                    Map<String, Object> result = ObjectPath.createFromResponse(resp).evaluate("_shards");
+                    assertThat(result.get("failed"), equalTo(0));
+                }
+            } catch (ResponseException ex) {
+                assertThat(ex.getResponse().getStatusLine(), equalTo(HttpStatus.SC_CONFLICT));
+                if (retryOnConflict) {
+                    throw new AssertionError(ex); // cause assert busy to retry
+                }
+            }
+        });
     }
 
     /**