Merge branch 'main' into ltr-min-window-size

afoucret · Jan 16, 2024 · 0c281b1 · 0c281b1
2 parents c3155bd + 51caf17
commit 0c281b1
Show file tree

Hide file tree

Showing 100 changed files with 413 additions and 683 deletions.
diff --git a/...tegTest/groovy/org/elasticsearch/gradle/internal/SymbolicLinkPreservingTarFuncTest.groovy b/...tegTest/groovy/org/elasticsearch/gradle/internal/SymbolicLinkPreservingTarFuncTest.groovy
@@ -36,6 +36,11 @@ class SymbolicLinkPreservingTarFuncTest extends AbstractGradleFuncTest {
         final Path linkToRealFolder = archiveSourceRoot.resolve("link-to-real-folder");
         Files.createSymbolicLink(linkToRealFolder, Paths.get("./real-folder"));
 
+        final Path realFolder2 = testProjectDir.getRoot().toPath().resolve("real-folder2")
+        final Path realFolderSub = realFolder2.resolve("sub")
+        Files.createDirectory(realFolder2);
+        Files.createDirectory(realFolderSub);
+
         buildFile << """
 import org.elasticsearch.gradle.internal.SymbolicLinkPreservingTar
 
@@ -56,6 +61,12 @@ tasks.register("buildBZip2Tar", SymbolicLinkPreservingTar) { SymbolicLinkPreserv
   tar.compression = Compression.BZIP2
   tar.preserveFileTimestamps = ${preserverTimestamp}
   from fileTree("archiveRoot")
+
+  into('config') {
+    dirMode 0750
+    fileMode 0660
+    from "real-folder2"
+  }
 }
 """
         when:
@@ -117,15 +128,22 @@ tasks.register("buildTar", SymbolicLinkPreservingTar) { SymbolicLinkPreservingTa
             while (entry != null) {
                 if (entry.getName().equals("real-folder/")) {
                     assert entry.isDirectory()
+                    assert entry.getMode() == 16877
                     realFolderEntry = true
-                } else if (entry.getName().equals("real-folder/file")) {
+                }  else if (entry.getName().equals("real-folder/file")) {
                     assert entry.isFile()
                     fileEntry = true
                 } else if (entry.getName().equals("real-folder/link-to-file")) {
                     assert entry.isSymbolicLink()
                     assert normalized(entry.getLinkName()) == "./file"
                     linkToFileEntry = true
-                } else if (entry.getName().equals("link-in-folder/")) {
+                } else if (entry.getName().equals("config/")) {
+                    assert entry.isDirectory()
+                    assert entry.getMode() == 16877
+                } else if (entry.getName().equals("config/sub/")) {
+                    assert entry.isDirectory()
+                    assert entry.getMode() == 16872
+                }else if (entry.getName().equals("link-in-folder/")) {
                     assert entry.isDirectory()
                     linkInFolderEntry = true
                 } else if (entry.getName().equals("link-in-folder/link-to-file")) {

diff --git a/...lusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java b/...lusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java
@@ -13,7 +13,6 @@
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsAction;
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
-import org.elasticsearch.action.admin.indices.flush.FlushResponse;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
@@ -30,6 +29,7 @@
 import org.elasticsearch.action.datastreams.lifecycle.ErrorEntry;
 import org.elasticsearch.action.datastreams.lifecycle.ExplainIndexDataStreamLifecycle;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.cluster.coordination.StableMasterHealthIndicatorService;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -313,7 +313,7 @@ public void testAutomaticForceMerge() throws Exception {
             for (int i = 0; i < randomIntBetween(10, 50); i++) {
                 indexDocs(dataStreamName, randomIntBetween(1, 300));
                 // Make sure the segments get written:
-                FlushResponse flushResponse = indicesAdmin().flush(new FlushRequest(toBeRolledOverIndex)).actionGet();
+                BroadcastResponse flushResponse = indicesAdmin().flush(new FlushRequest(toBeRolledOverIndex)).actionGet();
                 assertThat(flushResponse.getStatus(), equalTo(RestStatus.OK));
             }
 

diff --git a/...ams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/...ams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java
@@ -19,7 +19,6 @@
 import org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
-import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockAction;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockRequest;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse;
@@ -33,6 +32,7 @@
 import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.downsample.DownsampleConfig;
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
@@ -1168,7 +1168,7 @@ private void forceMergeIndex(ForceMergeRequest forceMergeRequest, ActionListener
         logger.info("Data stream lifecycle is issuing a request to force merge index [{}]", targetIndex);
         client.admin().indices().forceMerge(forceMergeRequest, new ActionListener<>() {
             @Override
-            public void onResponse(ForceMergeResponse forceMergeResponse) {
+            public void onResponse(BroadcastResponse forceMergeResponse) {
                 if (forceMergeResponse.getFailedShards() > 0) {
                     DefaultShardOperationFailedException[] failures = forceMergeResponse.getShardFailures();
                     String message = Strings.format(

diff --git a/...rc/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/...rc/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java
@@ -15,7 +15,6 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
-import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
 import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockRequest;
 import org.elasticsearch.action.admin.indices.rollover.MaxAgeCondition;
 import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
@@ -27,6 +26,7 @@
 import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.downsample.DownsampleConfig;
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
@@ -578,7 +578,7 @@ public void testForceMerge() throws Exception {
         // We want this test method to get fake force merge responses, because this is what triggers a cluster state update
         clientDelegate = (action, request, listener) -> {
             if (action.name().equals("indices:admin/forcemerge")) {
-                listener.onResponse(new ForceMergeResponse(5, 5, 0, List.of()));
+                listener.onResponse(new BroadcastResponse(5, 5, 0, List.of()));
             }
         };
         String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
@@ -748,7 +748,7 @@ public void testForceMergeRetries() throws Exception {
             clientDelegate = (action, request, listener) -> {
                 if (action.name().equals("indices:admin/forcemerge")) {
                     listener.onResponse(
-                        new ForceMergeResponse(
+                        new BroadcastResponse(
                             5,
                             5,
                             1,
@@ -779,7 +779,7 @@ public void testForceMergeRetries() throws Exception {
             AtomicInteger forceMergeFailedCount = new AtomicInteger(0);
             clientDelegate = (action, request, listener) -> {
                 if (action.name().equals("indices:admin/forcemerge")) {
-                    listener.onResponse(new ForceMergeResponse(5, 4, 0, List.of()));
+                    listener.onResponse(new BroadcastResponse(5, 4, 0, List.of()));
                     forceMergeFailedCount.incrementAndGet();
                 }
             };
@@ -800,7 +800,7 @@ public void testForceMergeRetries() throws Exception {
             // For the final data stream lifecycle run, we let forcemerge run normally
             clientDelegate = (action, request, listener) -> {
                 if (action.name().equals("indices:admin/forcemerge")) {
-                    listener.onResponse(new ForceMergeResponse(5, 5, 0, List.of()));
+                    listener.onResponse(new BroadcastResponse(5, 5, 0, List.of()));
                 }
             };
             dataStreamLifecycleService.run(clusterService.state());
@@ -900,7 +900,7 @@ public void testForceMergeDedup() throws Exception {
         setState(clusterService, state);
         clientDelegate = (action, request, listener) -> {
             if (action.name().equals("indices:admin/forcemerge")) {
-                listener.onResponse(new ForceMergeResponse(5, 5, 0, List.of()));
+                listener.onResponse(new BroadcastResponse(5, 5, 0, List.of()));
             }
         };
         for (int i = 0; i < 100; i++) {

diff --git a/modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTests.java b/modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTests.java
@@ -17,13 +17,12 @@
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.indices.flush.FlushAction;
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
-import org.elasticsearch.action.admin.indices.flush.FlushResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
-import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.index.TransportIndexAction;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlocks;
@@ -178,28 +177,34 @@ public int read() throws IOException {
     }
 
     public void testIndexChunksNoData() throws IOException {
-        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<FlushResponse> flushResponseActionListener) -> {
+        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
             assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(FlushResponse.class));
-        });
-        client.addHandler(RefreshAction.INSTANCE, (RefreshRequest request, ActionListener<RefreshResponse> flushResponseActionListener) -> {
-            assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(RefreshResponse.class));
+            flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
         });
+        client.addHandler(
+            RefreshAction.INSTANCE,
+            (RefreshRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
+                assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
+                flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
+            }
+        );
 
         InputStream empty = new ByteArrayInputStream(new byte[0]);
         assertEquals(0, geoIpDownloader.indexChunks("test", empty, 0, "d41d8cd98f00b204e9800998ecf8427e", 0));
     }
 
     public void testIndexChunksMd5Mismatch() {
-        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<FlushResponse> flushResponseActionListener) -> {
-            assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(FlushResponse.class));
-        });
-        client.addHandler(RefreshAction.INSTANCE, (RefreshRequest request, ActionListener<RefreshResponse> flushResponseActionListener) -> {
+        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
             assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(RefreshResponse.class));
+            flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
         });
+        client.addHandler(
+            RefreshAction.INSTANCE,
+            (RefreshRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
+                assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
+                flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
+            }
+        );
 
         IOException exception = expectThrows(
             IOException.class,
@@ -232,14 +237,17 @@ public void testIndexChunks() throws IOException {
             assertEquals(chunk + 15, source.get("chunk"));
             listener.onResponse(mock(IndexResponse.class));
         });
-        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<FlushResponse> flushResponseActionListener) -> {
-            assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(FlushResponse.class));
-        });
-        client.addHandler(RefreshAction.INSTANCE, (RefreshRequest request, ActionListener<RefreshResponse> flushResponseActionListener) -> {
+        client.addHandler(FlushAction.INSTANCE, (FlushRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
             assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
-            flushResponseActionListener.onResponse(mock(RefreshResponse.class));
+            flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
         });
+        client.addHandler(
+            RefreshAction.INSTANCE,
+            (RefreshRequest request, ActionListener<BroadcastResponse> flushResponseActionListener) -> {
+                assertArrayEquals(new String[] { GeoIpDownloader.DATABASES_INDEX }, request.indices());
+                flushResponseActionListener.onResponse(mock(BroadcastResponse.class));
+            }
+        );
 
         InputStream big = new ByteArrayInputStream(bigArray);
         assertEquals(17, geoIpDownloader.indexChunks("test", big, 15, "a67563dfa8f3cba8b8cff61eb989a749", 0));

diff --git a/modules/reindex/src/main/java/org/elasticsearch/reindex/AbstractAsyncBulkByScrollAction.java b/modules/reindex/src/main/java/org/elasticsearch/reindex/AbstractAsyncBulkByScrollAction.java
@@ -13,7 +13,6 @@
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
-import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.bulk.BackoffPolicy;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse.Failure;
@@ -24,6 +23,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.client.internal.ParentTaskAssigningClient;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
@@ -554,9 +554,9 @@ void refreshAndFinish(List<Failure> indexingFailures, List<SearchFailure> search
         RefreshRequest refresh = new RefreshRequest();
         refresh.indices(destinationIndices.toArray(new String[destinationIndices.size()]));
         logger.debug("[{}]: refreshing", task.getId());
-        bulkClient.admin().indices().refresh(refresh, new ActionListener<RefreshResponse>() {
+        bulkClient.admin().indices().refresh(refresh, new ActionListener<>() {
             @Override
-            public void onResponse(RefreshResponse response) {
+            public void onResponse(BroadcastResponse response) {
                 finishHim(null, indexingFailures, searchFailures, timedOut);
             }
 

diff --git a/...nternalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java b/...nternalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java
@@ -15,8 +15,8 @@
 import com.sun.net.httpserver.HttpHandler;
 
 import org.elasticsearch.action.ActionRunnable;
-import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -191,7 +191,7 @@ public void testAbortRequestStats() throws Exception {
             waitForDocs(nbDocs, indexer);
         }
         flushAndRefresh(index);
-        ForceMergeResponse forceMerge = client().admin().indices().prepareForceMerge(index).setFlush(true).setMaxNumSegments(1).get();
+        BroadcastResponse forceMerge = client().admin().indices().prepareForceMerge(index).setFlush(true).setMaxNumSegments(1).get();
         assertThat(forceMerge.getSuccessfulShards(), equalTo(1));
         assertHitCount(prepareSearch(index).setSize(0).setTrackTotalHits(true), nbDocs);
 
@@ -234,7 +234,7 @@ public void testMetrics() throws Exception {
             waitForDocs(nbDocs, indexer);
         }
         flushAndRefresh(index);
-        ForceMergeResponse forceMerge = client().admin().indices().prepareForceMerge(index).setFlush(true).setMaxNumSegments(1).get();
+        BroadcastResponse forceMerge = client().admin().indices().prepareForceMerge(index).setFlush(true).setMaxNumSegments(1).get();
         assertThat(forceMerge.getSuccessfulShards(), equalTo(1));
         assertHitCount(prepareSearch(index).setSize(0).setTrackTotalHits(true), nbDocs);
 

diff --git a/...luster-restart/src/javaRestTest/java/org/elasticsearch/upgrades/FullClusterRestartIT.java b/...luster-restart/src/javaRestTest/java/org/elasticsearch/upgrades/FullClusterRestartIT.java
@@ -605,7 +605,7 @@ public void testShrinkAfterUpgrade() throws IOException {
      *  <li>Make sure the document count is correct
      * </ol>
      */
-    public void testRollover() throws IOException {
+    public void testRollover() throws Exception {
         if (isRunningAgainstOldCluster()) {
             client().performRequest(
                 newXContentRequest(
@@ -637,9 +637,12 @@ public void testRollover() throws IOException {
                 )
             );
 
-            assertThat(
-                EntityUtils.toString(client().performRequest(new Request("GET", "/_cat/indices?v")).getEntity()),
-                containsString("testrollover-000002")
+            // assertBusy to work around https://github.com/elastic/elasticsearch/issues/104371
+            assertBusy(
+                () -> assertThat(
+                    EntityUtils.toString(client().performRequest(new Request("GET", "/_cat/indices?v&error_trace")).getEntity()),
+                    containsString("testrollover-000002")
+                )
             );
         }
 

diff --git a/qa/multi-cluster-search/src/test/java/org/elasticsearch/search/CCSDuelIT.java b/qa/multi-cluster-search/src/test/java/org/elasticsearch/search/CCSDuelIT.java
@@ -17,8 +17,8 @@
 import org.apache.lucene.search.join.ScoreMode;
 import org.apache.lucene.tests.util.TimeUnits;
 import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.aggregations.pipeline.DerivativePipelineAggregationBuilder;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
@@ -199,7 +199,7 @@ private void indexDocuments(String idPrefix) throws IOException, InterruptedExce
 
         assertTrue(latch.await(30, TimeUnit.SECONDS));
 
-        RefreshResponse refreshResponse = refresh(INDEX_NAME);
+        BroadcastResponse refreshResponse = refresh(INDEX_NAME);
         ElasticsearchAssertions.assertNoFailures(refreshResponse);
     }