Merge branch 'main' into popraw

Author: iverase

.buildkite/pipelines/periodic-fwc.template.yml

@@ -1,6 +1,6 @@
 steps:
-  - label: $FWC_VERSION / fwc
-    command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v$FWC_VERSION#fwcTest -Dtests.bwc.snapshot=false
+  - label: "{{matrix.FWC_VERSION}}" / fwc
+    command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v$$FWC_VERSION#fwcTest -Dtests.bwc.snapshot=false
     timeout_in_minutes: 300
     agents:
       provider: gcp
@@ -11,4 +11,4 @@ steps:
       setup:
         FWC_VERSION: $FWC_LIST
     env:
-      FWC_VERSION: $FWC_VERSION
+      FWC_VERSION: "{{matrix.FWC_VERSION}}"

.buildkite/pipelines/periodic-fwc.yml

@@ -1,7 +1,7 @@
 # This file is auto-generated. See .buildkite/pipelines/periodic-fwc.template.yml
 steps:
-  - label: $FWC_VERSION / fwc
-    command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v$FWC_VERSION#fwcTest -Dtests.bwc.snapshot=false
+  - label: "{{matrix.FWC_VERSION}}" / fwc
+    command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v$$FWC_VERSION#fwcTest -Dtests.bwc.snapshot=false
     timeout_in_minutes: 300
     agents:
       provider: gcp
@@ -12,4 +12,4 @@ steps:
       setup:
         FWC_VERSION: []
     env:
-      FWC_VERSION: $FWC_VERSION
+      FWC_VERSION: "{{matrix.FWC_VERSION}}"

.buildkite/scripts/fwc-branches.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+
+# Configure FwC test branches
+# We do not want 7.x branch and only to run for branches that:
+# - have released at least one minor version (not main)
+# - have previous minor unreleased (not the oldest development branch)
+FWC_BRANCHES=()
+for branch in "${BRANCHES[@]}"; do
+  if [[ ! "$branch" =~ ^7\..* ]]; then
+    FWC_BRANCHES+=("$branch")
+  fi
+done
+# Remove first and last element
+FWC_BRANCHES=("${FWC_BRANCHES[@]:1:${#FWC_BRANCHES[@]}-2}")
+
+shouldRunFwcFor() {
+  local branch=$1
+  for fwc_branch in "${FWC_BRANCHES[@]}"; do
+    if [[ "$fwc_branch" == "$branch" ]]; then
+      return 0
+    fi
+  done
+  return 1
+}

.buildkite/scripts/periodic.trigger.sh

@@ -5,6 +5,7 @@ set -euo pipefail
 echo "steps:"
 
 source .buildkite/scripts/branches.sh
+source .buildkite/scripts/fwc-branches.sh
 
 IS_FIRST=true
 SKIP_DELAY="${SKIP_DELAY:-false}"
@@ -46,8 +47,7 @@ EOF
       branch: "$BRANCH"
       commit: "$LAST_GOOD_COMMIT"
 EOF
-# Include forward compatibility tests only for the bugfix branch
-if [[ "${BRANCH}" == "${BRANCHES[2]}" ]]; then
+if shouldRunFwcFor "$BRANCH"; then
   cat <<EOF
   - trigger: elasticsearch-periodic-fwc
     label: Trigger periodic-fwc pipeline for $BRANCH

build-tools-internal/src/main/groovy/elasticsearch.fwc-test.gradle

@@ -7,19 +7,23 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-import org.elasticsearch.gradle.VersionProperties
+import org.elasticsearch.gradle.Version
 import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask
 
-def fwcVersions = buildParams.bwcVersions.released.findAll { it.major == VersionProperties.elasticsearchVersion.major &&  it.minor == VersionProperties.elasticsearchVersion.minor }
-def previousMinorSnapshot = buildParams.bwcVersions.unreleased.find { it.major == VersionProperties.elasticsearchVersion.major && it.minor == VersionProperties.elasticsearchVersion.minor - 1 }
-
-fwcVersions.each { fwcVersion ->
-  tasks.register("v${fwcVersion}#fwcTest", StandaloneRestIntegTestTask) {
-    usesBwcDistribution(previousMinorSnapshot)
-    usesBwcDistribution(fwcVersion)
-    systemProperty("tests.old_cluster_version", previousMinorSnapshot)
-    systemProperty("tests.new_cluster_version", fwcVersion)
-    nonInputProperties.systemProperty 'tests.fwc', 'true'
+Version elasticsearchVersion = Version.fromString(versions.get("elasticsearch"))
+def fwcVersions = buildParams.bwcVersions.released.findAll { it.major == elasticsearchVersion.major &&  it.minor == elasticsearchVersion.minor }
+def targetMajor = elasticsearchVersion.minor > 0 ? elasticsearchVersion.major : elasticsearchVersion.major - 1
+def targetMinor = elasticsearchVersion.minor > 0 ? elasticsearchVersion.minor - 1 : buildParams.bwcVersions.unreleased.findAll { it.major == targetMajor }*.minor.max()
+def previousMinorSnapshot = buildParams.bwcVersions.unreleased.find { it.major == targetMajor && it.minor == targetMinor }
+if (previousMinorSnapshot != null) {
+  fwcVersions.each { fwcVersion ->
+    tasks.register("v${fwcVersion}#fwcTest", StandaloneRestIntegTestTask) {
+      usesBwcDistribution(previousMinorSnapshot)
+      usesBwcDistribution(fwcVersion)
+      systemProperty("tests.old_cluster_version", previousMinorSnapshot)
+      systemProperty("tests.new_cluster_version", fwcVersion)
+      nonInputProperties.systemProperty 'tests.fwc', 'true'
+    }
   }
 }
 

docs/changelog/131940.yaml

@@ -0,0 +1,5 @@
+pr: 131940
+summary: Allow remote enrich after LOOKUP JOIN
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/133616.yaml

@@ -0,0 +1,5 @@
+pr: 133616
+summary: Add mode filter to _resolve/index
+area: Indices APIs
+type: enhancement
+issues: []

docs/changelog/133775.yaml

@@ -0,0 +1,5 @@
+pr: 133775
+summary: Remove Transfer-Encoding from HTTP request with no content
+area: Network
+type: bug
+issues: []

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java

@@ -85,6 +85,7 @@
 import org.elasticsearch.transport.netty4.Netty4Utils;
 import org.elasticsearch.xcontent.json.JsonXContent;
 
+import java.io.InputStream;
 import java.nio.channels.ClosedChannelException;
 import java.nio.charset.StandardCharsets;
 import java.util.Collection;
@@ -392,6 +393,23 @@ public void testOversizedChunkedEncoding() throws Exception {
         }
     }
 
+    public void testEmptyChunkedEncoding() throws Exception {
+        try (var clientContext = newClientContext()) {
+            var opaqueId = clientContext.newOpaqueId();
+            final var emptyStream = new HttpChunkedInput(new ChunkedStream(InputStream.nullInputStream()));
+            final var request = httpRequest(opaqueId, 0);
+            HttpUtil.setTransferEncodingChunked(request, true);
+            clientContext.channel().pipeline().addLast(new ChunkedWriteHandler());
+            clientContext.channel().writeAndFlush(request);
+            clientContext.channel().writeAndFlush(emptyStream);
+
+            var handler = clientContext.awaitRestChannelAccepted(opaqueId);
+            var restRequest = handler.restRequest;
+            assertFalse(restRequest.hasContent());
+            assertNull(restRequest.header("Transfer-Encoding"));
+        }
+    }
+
     // ensures that we don't leak buffers in stream on 400-bad-request
     // some bad requests are dispatched from rest-controller before reaching rest handler
     // test relies on netty's buffer leak detection
@@ -733,15 +751,17 @@ Channel channel() {
     static class ServerRequestHandler implements BaseRestHandler.RequestBodyChunkConsumer {
         final SubscribableListener<Void> channelAccepted = new SubscribableListener<>();
         final String opaqueId;
+        final RestRequest restRequest;
         private final AtomicReference<ActionListener<Chunk>> nextChunkListenerRef = new AtomicReference<>();
         final Netty4HttpRequestBodyStream stream;
         RestChannel channel;
         boolean receivedLastChunk = false;
         final CountDownLatch closedLatch = new CountDownLatch(1);
         volatile boolean shouldThrowInsideHandleChunk = false;
 
-        ServerRequestHandler(String opaqueId, Netty4HttpRequestBodyStream stream) {
+        ServerRequestHandler(String opaqueId, RestRequest restRequest, Netty4HttpRequestBodyStream stream) {
             this.opaqueId = opaqueId;
+            this.restRequest = restRequest;
             this.stream = stream;
         }
 
@@ -934,7 +954,7 @@ public List<Route> routes() {
                 protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
                     var stream = (Netty4HttpRequestBodyStream) request.contentStream();
                     var opaqueId = request.getHeaders().get(Task.X_OPAQUE_ID_HTTP_HEADER).get(0);
-                    var handler = new ServerRequestHandler(opaqueId, stream);
+                    var handler = new ServerRequestHandler(opaqueId, request, stream);
                     handlersByOpaqueId.getHandlerFor(opaqueId).onResponse(handler);
                     return handler;
                 }

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/transport/netty4/ESLoggingHandlerIT.java

@@ -10,11 +10,13 @@
 package org.elasticsearch.transport.netty4;
 
 import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
 import org.elasticsearch.ESNetty4IntegTestCase;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.transport.NodeDisconnectedException;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportLogger;
 
@@ -117,7 +119,15 @@ public void testExceptionalDisconnectLogging() throws Exception {
                 TcpTransport.class.getCanonicalName(),
                 Level.DEBUG,
                 ".*closed transport connection \\[[1-9][0-9]*\\] to .* with age \\[[0-9]+ms\\], exception:.*"
-            )
+            ) {
+                @Override
+                public void match(LogEvent event) {
+                    if (event.getThrown() instanceof NodeDisconnectedException nodeDisconnectedException
+                        && nodeDisconnectedException.getMessage().contains("closed exceptionally: Netty4TcpChannel{")) {
+                        super.match(event);
+                    }
+                }
+            }
         );
 
         final String nodeName = internalCluster().startNode();