Merge branch 'main' into generalExpressionJoin_v2

Author: julian-elastic

build-tools-internal/gradle/wrapper/gradle-wrapper.properties

@@ -1,7 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=16f2b95838c1ddcf7242b1c39e7bbbb43c842f1f1a1a0dc4959b6d4d68abcac3
-distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.0-all.zip
+distributionSha256Sum=f86344275d1b194688dd330abf9f6f2344cd02872ffee035f2d1ea2fd60cf7f3
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.1-all.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

build-tools-internal/src/main/resources/minimumGradleVersion

@@ -1 +1 @@
-9.2.0
+9.2.1

docs/changelog/138624.yaml

@@ -0,0 +1,6 @@
+pr: 138624
+summary: Handle individual doc parsing failure in bulk request with pipeline
+area: Ingest Node
+type: bug
+issues:
+ - 138445

gradle/wrapper/gradle-wrapper.properties

@@ -1,7 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=16f2b95838c1ddcf7242b1c39e7bbbb43c842f1f1a1a0dc4959b6d4d68abcac3
-distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.0-all.zip
+distributionSha256Sum=f86344275d1b194688dd330abf9f6f2344cd02872ffee035f2d1ea2fd60cf7f3
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.1-all.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/DerParser.java

@@ -138,7 +138,12 @@ private int getLength() throws IOException {
         int n = derInputStream.read(bytes);
         if (n < num) throw new IOException("Invalid DER: length too short");
 
-        return new BigInteger(1, bytes).intValue();
+        int len = new BigInteger(1, bytes).intValue();
+        if (len < 0) {
+            throw new IOException("Invalid DER: length larger than max-int");
+        }
+
+        return len;
     }
 
     /**

modules/reindex/src/main/java/org/elasticsearch/reindex/ReindexPlugin.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.DeleteByQueryAction;
@@ -44,6 +45,11 @@ public class ReindexPlugin extends Plugin implements ActionPlugin {
 
     public static final ActionType<ListTasksResponse> RETHROTTLE_ACTION = new ActionType<>("cluster:admin/reindex/rethrottle");
 
+    /**
+     * Whether the feature flag to guard the work to make reindex more resilient while it is under development.
+     */
+    static boolean REINDEX_RESILIENCE_ENABLED = new FeatureFlag("reindex_resilience").isEnabled();
+
     @Override
     public List<ActionHandler> getActions() {
         return Arrays.asList(

muted-tests.yml

@@ -60,8 +60,6 @@ tests:
 - class: org.elasticsearch.xpack.ccr.FollowIndexSecurityIT
   method: testCleanShardFollowTaskAfterDeleteFollower
   issue: https://github.com/elastic/elasticsearch/issues/120339
-- class: org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeToCharProcessorTests
-  issue: https://github.com/elastic/elasticsearch/issues/120575
 - class: org.elasticsearch.xpack.security.authc.service.ServiceAccountIT
   method: testAuthenticateShouldNotFallThroughInCaseOfFailure
   issue: https://github.com/elastic/elasticsearch/issues/120902
@@ -438,15 +436,15 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeForkIT
   method: test {csv-spec:inlinestats.MvMinMvExpand}
   issue: https://github.com/elastic/elasticsearch/issues/137679
-- class: org.elasticsearch.xpack.esql.optimizer.rules.physical.local.SubstituteRoundToTests
-  method: testSubqueryWithCountStarAndDateTrunc {default}
-  issue: https://github.com/elastic/elasticsearch/issues/138601
 - class: org.elasticsearch.xpack.ilm.TimeSeriesLifecycleActionsIT
   method: testWaitForSnapshot
   issue: https://github.com/elastic/elasticsearch/issues/138669
 - class: org.elasticsearch.test.rest.yaml.CcsCommonYamlTestSuiteIT
   method: test {p0=scroll/20_keep_alive/Max keep alive}
   issue: https://github.com/elastic/elasticsearch/issues/138680
+- class: org.elasticsearch.xpack.unsignedlong.UnsignedLongSyntheticSourceNativeArrayIntegrationTests
+  method: testSynthesizeArrayRandom
+  issue: https://github.com/elastic/elasticsearch/issues/138684
 
 # Examples:
 #

plugins/examples/gradle/wrapper/gradle-wrapper.properties

@@ -1,7 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=16f2b95838c1ddcf7242b1c39e7bbbb43c842f1f1a1a0dc4959b6d4d68abcac3
-distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.0-all.zip
+distributionSha256Sum=f86344275d1b194688dd330abf9f6f2344cd02872ffee035f2d1ea2fd60cf7f3
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.1-all.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

server/src/internalClusterTest/java/org/elasticsearch/ingest/IngestClientIT.java

@@ -378,6 +378,92 @@ public void testPipelineProcessorOnFailure() throws Exception {
         assertThat(inserted.get("readme"), equalTo("pipeline with id [3] is a bad pipeline"));
     }
 
+    public void testBulkRequestWithInvalidJsonAndPipeline() throws Exception {
+        // Test that when a document with invalid JSON is in a bulk request with a pipeline,
+        // the invalid document fails gracefully without causing the entire bulk request to fail.
+        // This tests the fix for https://github.com/elastic/elasticsearch/issues/138445
+
+        createIndex("test_index");
+
+        putJsonPipeline(
+            "test-pipeline",
+            (builder, params) -> builder.field("description", "test pipeline")
+                .startArray("processors")
+                .startObject()
+                .startObject("test")
+                .endObject()
+                .endObject()
+                .endArray()
+        );
+
+        // Create a bulk request with valid and invalid documents
+        BulkRequest bulkRequest = new BulkRequest();
+
+        // Valid document
+        IndexRequest validRequest = new IndexRequest("test_index").id("valid_doc");
+        validRequest.source("{\"valid\":\"test\"}", XContentType.JSON);
+        validRequest.setPipeline("test-pipeline");
+        bulkRequest.add(validRequest);
+
+        // Invalid document with missing closing brace
+        IndexRequest invalidRequest = new IndexRequest("test_index").id("invalid_doc");
+        invalidRequest.source("{\"invalid\":\"json\"", XContentType.JSON);
+        invalidRequest.setPipeline("test-pipeline");
+        bulkRequest.add(invalidRequest);
+
+        // Invalid document with duplicate fields
+        IndexRequest invalidRequest2 = new IndexRequest("test_index").id("invalid_doc2");
+        invalidRequest2.source("{\"invalid\":\"json\", \"invalid\":\"json\"}", XContentType.JSON);
+        invalidRequest2.setPipeline("test-pipeline");
+        bulkRequest.add(invalidRequest2);
+
+        // Another valid document
+        IndexRequest validRequest2 = new IndexRequest("test_index").id("valid_doc2");
+        validRequest2.source("{\"valid\":\"test2\"}", XContentType.JSON);
+        validRequest2.setPipeline("test-pipeline");
+        bulkRequest.add(validRequest2);
+
+        BulkResponse response = client().bulk(bulkRequest).actionGet();
+
+        // The bulk request should succeed
+        assertThat(response.hasFailures(), is(true));
+        assertThat(response.getItems().length, equalTo(4));
+
+        // First document should succeed
+        BulkItemResponse item0 = response.getItems()[0];
+        assertThat(item0.isFailed(), is(false));
+        assertThat(item0.getResponse().getId(), equalTo("valid_doc"));
+        assertThat(item0.getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+
+        // Second document should fail
+        BulkItemResponse item1 = response.getItems()[1];
+        assertThat(item1.isFailed(), is(true));
+        assertThat(item1.getFailure().getStatus(), equalTo(org.elasticsearch.rest.RestStatus.BAD_REQUEST));
+        assertThat(item1.getFailure().getCause(), instanceOf(IllegalArgumentException.class));
+
+        // Third document should fail
+        BulkItemResponse item2 = response.getItems()[2];
+        assertThat(item2.isFailed(), is(true));
+        assertThat(item2.getFailure().getStatus(), equalTo(org.elasticsearch.rest.RestStatus.BAD_REQUEST));
+        assertThat(item2.getFailure().getCause(), instanceOf(IllegalArgumentException.class));
+
+        // Fourth document should succeed
+        BulkItemResponse item3 = response.getItems()[3];
+        assertThat(item3.isFailed(), is(false));
+        assertThat(item3.getResponse().getId(), equalTo("valid_doc2"));
+        assertThat(item3.getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+
+        // Verify that the valid documents were indexed
+        assertThat(client().prepareGet("test_index", "valid_doc").get().isExists(), is(true));
+        assertThat(client().prepareGet("test_index", "valid_doc2").get().isExists(), is(true));
+        // Verify that the invalid documents were not indexed
+        assertThat(client().prepareGet("test_index", "invalid_doc").get().isExists(), is(false));
+        assertThat(client().prepareGet("test_index", "invalid_doc2").get().isExists(), is(false));
+
+        // cleanup
+        deletePipeline("test-pipeline");
+    }
+
     public static class ExtendedIngestTestPlugin extends IngestTestPlugin {
 
         @Override

server/src/main/java/org/elasticsearch/cluster/ClusterModule.java

@@ -58,6 +58,7 @@
 import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.FilterAllocationDecider;
+import org.elasticsearch.cluster.routing.allocation.decider.IndexBalanceAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.IndexVersionAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.NodeReplacementAllocationDecider;
@@ -497,6 +498,7 @@ public static Collection<AllocationDecider> createAllocationDeciders(
         addAllocationDecider(deciders, new ThrottlingAllocationDecider(clusterSettings));
         addAllocationDecider(deciders, new ShardsLimitAllocationDecider(clusterSettings));
         addAllocationDecider(deciders, new AwarenessAllocationDecider(settings, clusterSettings));
+        addAllocationDecider(deciders, new IndexBalanceAllocationDecider(settings, clusterSettings));
 
         clusterPlugins.stream()
             .flatMap(p -> p.createAllocationDeciders(settings, clusterSettings).stream())