TransportVerifyShardBeforeCloseAction should force a flush (#38401)

This commit changes the `TransportVerifyShardBeforeCloseAction` so that it 
always forces the flush of the shard. It seems that #37961 is not sufficient to 
ensure that the translog and the Lucene commit share the exact same max 
seq no and global checkpoint information in case of one or more noop 
operations have been made.

The `BulkWithUpdatesIT.testThatMissingIndexDoesNotAbortFullBulkRequest` 
and `FrozenIndexTests.testFreezeEmptyIndexWithTranslogOps` test this trivial 
situation and they both fail 1 on 10 executions.

Relates to #33888

plugins/repository-hdfs/src/test/resources/rest-api-spec/test/hdfs_repository/40_restore.yml

@@ -62,10 +62,10 @@
 
   - match: { test_index.shards.0.type: SNAPSHOT }
   - match: { test_index.shards.0.stage: DONE }
-  - match: { test_index.shards.0.index.files.recovered: 0}
-  - match: { test_index.shards.0.index.size.recovered_in_bytes: 0}
-  - match: { test_index.shards.0.index.files.reused: 1}
-  - gt: { test_index.shards.0.index.size.reused_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.recovered: 1}
+  - gt:    { test_index.shards.0.index.size.recovered_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.reused: 0}
+  - match: { test_index.shards.0.index.size.reused_in_bytes: 0}
 
   # Remove our snapshot
   - do:

plugins/repository-hdfs/src/test/resources/rest-api-spec/test/secure_hdfs_repository/40_restore.yml

@@ -64,10 +64,10 @@
 
   - match: { test_index.shards.0.type: SNAPSHOT }
   - match: { test_index.shards.0.stage: DONE }
-  - match: { test_index.shards.0.index.files.recovered: 0}
-  - match: { test_index.shards.0.index.size.recovered_in_bytes: 0}
-  - match: { test_index.shards.0.index.files.reused: 1}
-  - gt: { test_index.shards.0.index.size.reused_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.recovered: 1}
+  - gt:    { test_index.shards.0.index.size.recovered_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.reused: 0}
+  - match: { test_index.shards.0.index.size.reused_in_bytes: 0}
 
   # Remove our snapshot
   - do:

rest-api-spec/src/main/resources/rest-api-spec/test/snapshot.restore/10_basic.yml

@@ -53,7 +53,7 @@ setup:
 
   - match: { test_index.shards.0.type: SNAPSHOT }
   - match: { test_index.shards.0.stage: DONE }
-  - match: { test_index.shards.0.index.files.recovered: 0}
-  - match: { test_index.shards.0.index.size.recovered_in_bytes: 0}
-  - match: { test_index.shards.0.index.files.reused: 1}
-  - gt: { test_index.shards.0.index.size.reused_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.recovered: 1}
+  - gt:    { test_index.shards.0.index.size.recovered_in_bytes: 0}
+  - match: { test_index.shards.0.index.files.reused: 0}
+  - match: { test_index.shards.0.index.size.reused_in_bytes: 0}

server/src/main/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseAction.java

@@ -115,9 +115,8 @@ private void executeShardOperation(final ShardRequest request, final IndexShard
                 + "] mismatches maximum sequence number [" + maxSeqNo + "] on index shard " + shardId);
         }
 
-        final boolean forced = indexShard.isSyncNeeded();
-        indexShard.flush(new FlushRequest().force(forced));
-        logger.trace("{} shard is ready for closing [forced:{}]", shardId, forced);
+        indexShard.flush(new FlushRequest().force(true));
+        logger.trace("{} shard is ready for closing", shardId);
     }
 
     @Override

server/src/test/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseActionTests.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.IndexShard;
@@ -56,6 +57,7 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.mockito.ArgumentCaptor;
 
 import java.util.Collections;
 import java.util.List;
@@ -69,6 +71,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -144,9 +147,13 @@ private void executeOnPrimaryOrReplica() throws Exception {
         }
     }
 
-    public void testOperationSuccessful() throws Exception {
+    public void testShardIsFlushed() throws Exception {
+        final ArgumentCaptor<FlushRequest> flushRequest = ArgumentCaptor.forClass(FlushRequest.class);
+        when(indexShard.flush(flushRequest.capture())).thenReturn(new Engine.CommitId(new byte[0]));
+
         executeOnPrimaryOrReplica();
         verify(indexShard, times(1)).flush(any(FlushRequest.class));
+        assertThat(flushRequest.getValue().force(), is(true));
     }
 
     public void testOperationFailsWithOnGoingOps() {