Merge branch 'main' into esql/constant_keyword_optimizations

Author: luigidellaquila

docs/changelog/127522.yaml

@@ -0,0 +1,6 @@
+pr: 127522
+summary: Use INTERNAL_INGEST for Inference
+area: Machine Learning
+type: bug
+issues:
+ - 127519

muted-tests.yml

@@ -429,18 +429,9 @@ tests:
 - class: org.elasticsearch.xpack.esql.heap_attack.HeapAttackIT
   method: testLookupExplosionNoFetch
   issue: https://github.com/elastic/elasticsearch/issues/127365
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testNonLocalRequestAlwaysFindsMasterAndWaitsForMetadata
-  issue: https://github.com/elastic/elasticsearch/issues/127422
 - class: org.elasticsearch.xpack.esql.qa.single_node.PushQueriesIT
   method: testPushCaseInsensitiveEqualityOnDefaults
   issue: https://github.com/elastic/elasticsearch/issues/127431
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testLocalRequestAlwaysSucceeds
-  issue: https://github.com/elastic/elasticsearch/issues/127423
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testLocalRequestWaitsForMetadata
-  issue: https://github.com/elastic/elasticsearch/issues/127466
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
   method: test
   issue: https://github.com/elastic/elasticsearch/issues/127536

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateActionDisruptionIT.java

@@ -28,17 +28,16 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.not;
 
 @ESIntegTestCase.ClusterScope(numDataNodes = 0, scope = ESIntegTestCase.Scope.TEST)
 public class TransportClusterStateActionDisruptionIT extends ESIntegTestCase {
@@ -212,11 +211,12 @@ public void runRepeatedlyWhileChangingMaster(Runnable runnable) throws Exception
             }
         }
 
-        assertBusy(() -> {
-            final String nonMasterNode = randomValueOtherThan(masterName, () -> randomFrom(internalCluster().getNodeNames()));
-            final String claimedMasterName = internalCluster().getMasterName(nonMasterNode);
-            assertThat(claimedMasterName, not(equalTo(masterName)));
-        });
+        final String nonMasterNode = randomValueOtherThan(masterName, () -> randomFrom(internalCluster().getNodeNames()));
+        awaitClusterState(
+            logger,
+            nonMasterNode,
+            state -> Optional.ofNullable(state.nodes().getMasterNode()).map(m -> m.getName().equals(masterName) == false).orElse(false)
+        );
 
         shutdown.set(true);
         assertingThread.join();

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilter.java

@@ -433,7 +433,15 @@ public void onFailure(Exception exc) {
                 }
             };
             inferenceProvider.service()
-                .chunkedInfer(inferenceProvider.model(), null, inputs, Map.of(), InputType.INGEST, TimeValue.MAX_VALUE, completionListener);
+                .chunkedInfer(
+                    inferenceProvider.model(),
+                    null,
+                    inputs,
+                    Map.of(),
+                    InputType.INTERNAL_INGEST,
+                    TimeValue.MAX_VALUE,
+                    completionListener
+                );
         }
 
         /**