Merge branch 'main' into add_microsoft_graph_plugin

jfreden · web-flow · commit 32c42840e170 · 2025-05-19T11:47:52.000+02:00
diff --git a/docs/changelog/128025.yaml b/docs/changelog/128025.yaml
@@ -0,0 +1,6 @@
+pr: 128025
+summary: "Set `connection: close` header on shutdown"
+area: Network
+type: enhancement
+issues:
+ - 127984
diff --git a/modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/http/netty4/Netty4PipeliningIT.java b/modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/http/netty4/Netty4PipeliningIT.java
@@ -9,11 +9,26 @@
 
 package org.elasticsearch.http.netty4;
 
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpVersion;
 import io.netty.util.ReferenceCounted;
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ESNetty4IntegTestCase;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.support.CountDownActionListener;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.internal.node.NodeClient;
@@ -29,6 +44,8 @@
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.http.HttpServerTransport;
@@ -41,29 +58,42 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.rest.action.EmptyResponseListener;
 import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.transport.netty4.NettyAllocator;
 import org.elasticsearch.xcontent.ToXContentObject;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_PIPELINING_MAX_EVENTS;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.oneOf;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
 public class Netty4PipeliningIT extends ESNetty4IntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return CollectionUtils.concatLists(List.of(CountDown3Plugin.class, ChunkAndFailPlugin.class), super.nodePlugins());
+        return CollectionUtils.concatLists(
+            List.of(CountDown3Plugin.class, ChunkAndFailPlugin.class, KeepPipeliningPlugin.class),
+            super.nodePlugins()
+        );
     }
 
     private static final int MAX_PIPELINE_EVENTS = 10;
@@ -142,6 +172,115 @@ private void runPipeliningTest(int expectedResponseCount, String... routes) thro
         }
     }
 
+    public void testSetCloseConnectionHeaderWhenShuttingDown() throws IOException {
+
+        // This test works using KeepPipeliningPlugin to keep a HTTP connection from becoming idle with a sequence of requests while the
+        // node shuts down and ensures that these requests start to receive responses with `Connection: close` and that the node does not
+        // shut down until all requests have received a response.
+
+        final var victimNode = internalCluster().startNode();
+
+        final var releasables = new ArrayList<Releasable>(3);
+        try {
+            final var keepPipeliningRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, KeepPipeliningPlugin.ROUTE);
+            releasables.add(keepPipeliningRequest::release);
+
+            final var enoughResponsesToCloseLatch = new CountDownLatch(between(1, 5));
+            final var outstandingRequestsCounter = new AtomicInteger();
+            final var nodeShuttingDown = new AtomicBoolean();
+            final var stoppedPipelining = new AtomicBoolean();
+
+            final EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
+            releasables.add(() -> eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS).awaitUninterruptibly());
+            final var clientBootstrap = new Bootstrap().channel(NettyAllocator.getChannelType())
+                .option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator())
+                .group(eventLoopGroup)
+                .handler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    protected void initChannel(SocketChannel ch) {
+                        ch.pipeline().addLast(new HttpClientCodec());
+                        ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpResponse>() {
+
+                            private int closeHeadersToIgnore = between(0, 5);
+
+                            private boolean ignoreCloseHeader() {
+                                if (closeHeadersToIgnore == 0) {
+                                    return false;
+                                } else {
+                                    closeHeadersToIgnore -= 1;
+                                    return true;
+                                }
+                            }
+
+                            @Override
+                            protected void channelRead0(ChannelHandlerContext ctx, HttpResponse msg) {
+                                enoughResponsesToCloseLatch.countDown();
+                                assertThat(
+                                    outstandingRequestsCounter.decrementAndGet(),
+                                    stoppedPipelining.get() ? oneOf(0, 1) : equalTo(1)
+                                );
+
+                                if ("close".equals(msg.headers().get("connection")) && ignoreCloseHeader() == false) {
+                                    assertTrue(nodeShuttingDown.get());
+                                    // send one more request with `?respond_immediately` to stop the pipelining
+                                    if (stoppedPipelining.compareAndSet(false, true)) {
+                                        assertThat(outstandingRequestsCounter.incrementAndGet(), equalTo(2));
+                                        ctx.writeAndFlush(
+                                            new DefaultFullHttpRequest(
+                                                HttpVersion.HTTP_1_1,
+                                                HttpMethod.GET,
+                                                KeepPipeliningPlugin.ROUTE + "?" + KeepPipeliningPlugin.RESPOND_IMMEDIATELY
+                                            )
+                                        );
+                                    }
+                                } else {
+                                    // still pipelining, send another request to trigger the next response
+                                    assertThat(outstandingRequestsCounter.incrementAndGet(), equalTo(2));
+                                    ctx.writeAndFlush(keepPipeliningRequest.retain());
+                                }
+                            }
+
+                            @Override
+                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+                                ExceptionsHelper.maybeDieOnAnotherThread(new AssertionError(cause));
+                            }
+                        });
+                    }
+                });
+
+            final var httpServerTransport = internalCluster().getInstance(HttpServerTransport.class, victimNode);
+            final var httpServerAddress = randomFrom(httpServerTransport.boundAddress().boundAddresses()).address();
+
+            // Open a channel on which we will pipeline the requests to KeepPipeliningPlugin.ROUTE
+            final var pipeliningChannel = clientBootstrap.connect(httpServerAddress).syncUninterruptibly().channel();
+            releasables.add(() -> pipeliningChannel.close().syncUninterruptibly());
+
+            // Send two pipelined requests so that we start to receive responses
+            assertTrue(outstandingRequestsCounter.compareAndSet(0, 2));
+            pipeliningChannel.writeAndFlush(keepPipeliningRequest.retain());
+            pipeliningChannel.writeAndFlush(keepPipeliningRequest.retain());
+
+            // wait until we've started to receive responses
+            safeAwait(enoughResponsesToCloseLatch);
+
+            // Shut down the node
+            assertTrue(nodeShuttingDown.compareAndSet(false, true));
+            internalCluster().stopNode(victimNode);
+
+            // Wait for the pipelining channel to be closed, indicating that it stopped pipelining (because it received a response with
+            // `Connection: close`) and allowed the node to shut down
+            pipeliningChannel.closeFuture().syncUninterruptibly();
+
+            // The shutdown did not happen until all requests had had a response.
+            assertTrue(stoppedPipelining.get());
+            assertEquals(0, outstandingRequestsCounter.get());
+
+        } finally {
+            Collections.reverse(releasables);
+            Releasables.close(releasables);
+        }
+    }
+
     private void assertOpaqueIdsInOrder(Collection<String> opaqueIds) {
         // check if opaque ids are monotonically increasing
         int i = 0;
@@ -203,7 +342,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
     }
 
     /**
-     * Adds an HTTP route that waits for 3 concurrent executions before returning any of them
+     * Adds an HTTP route that starts to emit a chunked response and then fails before its completion.
      */
     public static class ChunkAndFailPlugin extends Plugin implements ActionPlugin {
 
@@ -285,4 +424,54 @@ public String getResponseContentTypeString() {
             });
         }
     }
+
+    /**
+     * Adds an HTTP route that only responds when starting to process a second request, ensuring that there is always at least one in-flight
+     * request in the pipeline which keeps a connection from becoming idle.
+     */
+    public static class KeepPipeliningPlugin extends Plugin implements ActionPlugin {
+
+        static final String ROUTE = "/_test/keep_pipelining";
+        static final String RESPOND_IMMEDIATELY = "respond_immediately";
+
+        @Override
+        public Collection<RestHandler> getRestHandlers(
+            Settings settings,
+            NamedWriteableRegistry namedWriteableRegistry,
+            RestController restController,
+            ClusterSettings clusterSettings,
+            IndexScopedSettings indexScopedSettings,
+            SettingsFilter settingsFilter,
+            IndexNameExpressionResolver indexNameExpressionResolver,
+            Supplier<DiscoveryNodes> nodesInCluster,
+            Predicate<NodeFeature> clusterSupportsFeature
+        ) {
+            return List.of(new BaseRestHandler() {
+
+                private SubscribableListener<Void> lastRequestTrigger = new SubscribableListener<>();
+
+                @Override
+                public String getName() {
+                    return ROUTE;
+                }
+
+                @Override
+                public List<Route> routes() {
+                    return List.of(new Route(GET, ROUTE));
+                }
+
+                @Override
+                protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
+                    final var respondImmediately = request.paramAsBoolean(RESPOND_IMMEDIATELY, false);
+                    return channel -> {
+                        // all happens on a single thread in these tests, no need for concurrency protection
+                        final var previousRequestTrigger = lastRequestTrigger;
+                        lastRequestTrigger = respondImmediately ? SubscribableListener.nullSuccess() : new SubscribableListener<>();
+                        lastRequestTrigger.addListener(new EmptyResponseListener(channel).map(ignored -> ActionResponse.Empty.INSTANCE));
+                        previousRequestTrigger.onResponse(null);
+                    };
+                }
+            });
+        }
+    }
 }
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -447,36 +447,42 @@ tests:
 - class: org.elasticsearch.indices.stats.IndexStatsIT
   method: testThrottleStats
   issue: https://github.com/elastic/elasticsearch/issues/126359
-- class: org.elasticsearch.packaging.test.DockerTests
-  method: test040JavaUsesTheOsProvidedKeystore
-  issue: https://github.com/elastic/elasticsearch/issues/127437
+- class: org.elasticsearch.search.vectors.IVFKnnFloatVectorQueryTests
+  method: testRandomWithFilter
+  issue: https://github.com/elastic/elasticsearch/issues/127963
+- class: org.elasticsearch.search.vectors.IVFKnnFloatVectorQueryTests
+  method: testSearchBoost
+  issue: https://github.com/elastic/elasticsearch/issues/127969
+- class: org.elasticsearch.search.vectors.IVFKnnFloatVectorQueryTests
+  method: testFindFewer
+  issue: https://github.com/elastic/elasticsearch/issues/128002
 - class: org.elasticsearch.xpack.remotecluster.RemoteClusterSecurityRestIT
   method: testTaskCancellation
   issue: https://github.com/elastic/elasticsearch/issues/128009
 - class: org.elasticsearch.xpack.esql.action.CrossClusterQueryWithPartialResultsIT
   method: testOneRemoteClusterPartial
   issue: https://github.com/elastic/elasticsearch/issues/124055
-- class: org.elasticsearch.packaging.test.DockerTests
-  method: test041AmazonCaCertsAreInTheKeystore
-  issue: https://github.com/elastic/elasticsearch/issues/128007
 - class: org.elasticsearch.compute.aggregation.SampleDoubleAggregatorFunctionTests
   method: testSimpleWithCranky
   issue: https://github.com/elastic/elasticsearch/issues/128024
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
   method: test {lookup-join.MvJoinKeyOnTheLookupIndex ASYNC}
   issue: https://github.com/elastic/elasticsearch/issues/128030
+- class: org.elasticsearch.compute.aggregation.PercentileIntGroupingAggregatorFunctionTests
+  method: testManyInitialManyPartialFinalRunner
+  issue: https://github.com/elastic/elasticsearch/issues/128092
+- class: org.elasticsearch.packaging.test.DockerTests
+  method: test042KeystorePermissionsAreCorrect
+  issue: https://github.com/elastic/elasticsearch/issues/128018
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test072RunEsAsDifferentUserAndGroup
   issue: https://github.com/elastic/elasticsearch/issues/128031
 - class: org.elasticsearch.packaging.test.DockerTests
-  method: test042KeystorePermissionsAreCorrect
-  issue: https://github.com/elastic/elasticsearch/issues/128019
+  method: test122CanUseDockerLoggingConfig
+  issue: https://github.com/elastic/elasticsearch/issues/128110
 - class: org.elasticsearch.packaging.test.DockerTests
-  method: test073RunEsAsDifferentUserAndGroupWithoutBindMounting
-  issue: https://github.com/elastic/elasticsearch/issues/128044
-- class: org.elasticsearch.compute.aggregation.PercentileIntGroupingAggregatorFunctionTests
-  method: testManyInitialManyPartialFinalRunner
-  issue: https://github.com/elastic/elasticsearch/issues/128092
+  method: test041AmazonCaCertsAreInTheKeystore
+  issue: https://github.com/elastic/elasticsearch/issues/128006
 
 # Examples:
 #
diff --git a/qa/packaging/src/test/java/org/elasticsearch/packaging/util/docker/Docker.java b/qa/packaging/src/test/java/org/elasticsearch/packaging/util/docker/Docker.java
@@ -75,6 +75,13 @@ public class Docker {
     public static final int STARTUP_SLEEP_INTERVAL_MILLISECONDS = 1000;
     public static final int STARTUP_ATTEMPTS_MAX = 30;
 
+    private static final String ELASTICSEARCH_FULL_CLASSNAME = "org.elasticsearch.bootstrap.Elasticsearch";
+    private static final String FIND_ELASTICSEARCH_PROCESS = "for pid in $(ps -eo pid,comm | grep java | awk '\\''{print $1}'\\''); "
+        + "do cmdline=$(tr \"\\0\" \" \" < /proc/$pid/cmdline 2>/dev/null); [[ $cmdline == *"
+        + ELASTICSEARCH_FULL_CLASSNAME
+        + "* ]] && echo \"$pid: $cmdline\"; done";
+    // The length of the command exceeds what we can use for COLUMNS so we use a pipe to detect the process we're looking for
+
     /**
      * Tracks the currently running Docker image. An earlier implementation used a fixed container name,
      * but that appeared to cause problems with repeatedly destroying and recreating containers with
@@ -185,11 +192,8 @@ public static void waitForElasticsearchToStart() {
             try {
                 // Give the container enough time for security auto-configuration or a chance to crash out
                 Thread.sleep(STARTUP_SLEEP_INTERVAL_MILLISECONDS);
-
-                // Set COLUMNS so that `ps` doesn't truncate its output
-                psOutput = dockerShell.run("bash -c 'COLUMNS=4000 ps ax'").stdout();
-
-                if (psOutput.contains("org.elasticsearch.bootstrap.Elasticsearch")) {
+                psOutput = dockerShell.run("bash -c '" + FIND_ELASTICSEARCH_PROCESS + " | wc -l'").stdout();
+                if (psOutput.contains("1")) {
                     isElasticsearchRunning = true;
                     break;
                 }
diff --git a/server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java b/server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java
diff --git a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
