Merge branch 'main' into api-remove-index-block

Author: nielsbauman

distribution/tools/server-cli/src/test/java/org/elasticsearch/server/cli/MachineDependentHeapTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.server.cli;
 
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.test.ESTestCase;
 import org.hamcrest.Matcher;
 
@@ -55,7 +56,8 @@ public void testMasterOnlyOptions() throws Exception {
         assertHeapOptions(64, containsInAnyOrder("-Xmx31744m", "-Xms31744m"), "master");
     }
 
-    public void testMlOnlyOptions() throws Exception {
+    public void testMlOnlyOptions_new() throws Exception {
+        assumeTrue("feature flag must be enabled for new memory computation", new FeatureFlag("new_ml_memory_computation").isEnabled());
         assertHeapOptions(1, containsInAnyOrder("-Xmx272m", "-Xms272m"), "ml");
         assertHeapOptions(4, containsInAnyOrder("-Xmx1092m", "-Xms1092m"), "ml");
         assertHeapOptions(32, containsInAnyOrder("-Xmx5460m", "-Xms5460m"), "ml");
@@ -65,6 +67,20 @@ public void testMlOnlyOptions() throws Exception {
         assertHeapOptions(263, containsInAnyOrder("-Xmx21228m", "-Xms21228m"), "ml");
     }
 
+    public void testMlOnlyOptions_old() throws Exception {
+        assumeTrue(
+            "feature flag must be disabled for old memory computation",
+            new FeatureFlag("new_ml_memory_computation").isEnabled() == false
+        );
+        assertHeapOptions(1, containsInAnyOrder("-Xmx408m", "-Xms408m"), "ml");
+        assertHeapOptions(4, containsInAnyOrder("-Xmx1636m", "-Xms1636m"), "ml");
+        assertHeapOptions(32, containsInAnyOrder("-Xmx8192m", "-Xms8192m"), "ml");
+        assertHeapOptions(64, containsInAnyOrder("-Xmx11468m", "-Xms11468m"), "ml");
+        // We'd never see a node this big in Cloud, but this assertion proves that the 31GB absolute maximum
+        // eventually kicks in (because 0.4 * 16 + 0.1 * (263 - 16) > 31)
+        assertHeapOptions(263, containsInAnyOrder("-Xmx31744m", "-Xms31744m"), "ml");
+    }
+
     public void testDataNodeOptions() throws Exception {
         assertHeapOptions(1, containsInAnyOrder("-Xmx512m", "-Xms512m"), "data");
         assertHeapOptions(8, containsInAnyOrder("-Xmx4096m", "-Xms4096m"), "data");

docs/changelog/129302.yaml

@@ -0,0 +1,6 @@
+pr: 129302
+summary: Move HTTP content aggregation from Netty into `RestController`
+area: Network
+type: enhancement
+issues:
+ - 120746

docs/changelog/129606.yaml

@@ -0,0 +1,35 @@
+pr: 129606
+summary: Release FORK in tech preview
+area: ES|QL
+type: feature
+issues: []
+highlight:
+  title: Release FORK in tech preview
+  body: |-
+    Fork is a foundational building block that allows multiple branches of execution.
+    Conceptually, fork is:
+    - a bifurcation of the stream, with all data going to each fork branch, followed by
+    - a merge of the branches, enhanced with a discriminator column called FORK:
+
+    Example:
+
+    [source,yaml]
+    ----------------------------
+    FROM test
+    | FORK
+    ( WHERE content:"fox" )
+    ( WHERE content:"dog" )
+    | SORT _fork
+    ----------------------------
+
+    The FORK command add a discriminator column called `_fork`:
+
+    [source,yaml]
+    ----------------------------
+    | id  | content   | _fork |
+    |-----|-----------|-------|
+    | 3   | brown fox | fork1 |
+    | 4   | white dog | fork2 |
+    ----------------------------
+
+  notable: true

docs/reference/query-languages/esql/_snippets/functions/layout/categorize.md

@@ -1,2 +1,2 @@
 * [`BUCKET`](../../functions-operators/grouping-functions.md#esql-bucket)
-* [preview] [`CATEGORIZE`](../../functions-operators/grouping-functions.md#esql-categorize)
+* [`CATEGORIZE`](../../functions-operators/grouping-functions.md#esql-categorize)

docs/reference/query-languages/esql/_snippets/lists/grouping-functions.md

@@ -17,6 +17,7 @@
 import org.elasticsearch.ingest.geoip.stats.CacheStats;
 import org.elasticsearch.test.ESTestCase;
 
+import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
@@ -140,23 +141,25 @@ public void testPurgeCacheEntriesForDatabase() {
         GeoIpCache cache = new GeoIpCache(100);
         ProjectId projectId1 = randomUniqueProjectId();
         ProjectId projectId2 = randomUniqueProjectId();
-        String databasePath1 = "path/to/db1";
-        String databasePath2 = "path/to/db2";
+        // Turn the path strings into Paths to ensure that we always use the canonical string representation (this string literal does not
+        // round-trip when converting to a Path and back again on Windows):
+        Path databasePath1 = PathUtils.get("path/to/db1");
+        Path databasePath2 = PathUtils.get("path/to/db2");
         String ip1 = "127.0.0.1";
         String ip2 = "127.0.0.2";
 
         AbstractResponse response = mock(AbstractResponse.class);
-        cache.putIfAbsent(projectId1, ip1, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId1, ip2, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId2, ip1, databasePath1, ip -> response); // cache miss
-        cache.putIfAbsent(projectId1, ip1, databasePath2, ip -> response); // cache miss
-        cache.purgeCacheEntriesForDatabase(projectId1, PathUtils.get(databasePath1));
+        cache.putIfAbsent(projectId1, ip1, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId1, ip2, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId2, ip1, databasePath1.toString(), ip -> response); // cache miss
+        cache.putIfAbsent(projectId1, ip1, databasePath2.toString(), ip -> response); // cache miss
+        cache.purgeCacheEntriesForDatabase(projectId1, databasePath1);
         // should have purged entries for projectId1 and databasePath1...
-        assertNull(cache.get(projectId1, ip1, databasePath1));
-        assertNull(cache.get(projectId1, ip2, databasePath1));
+        assertNull(cache.get(projectId1, ip1, databasePath1.toString()));
+        assertNull(cache.get(projectId1, ip2, databasePath1.toString()));
         // ...but left the one for projectId2...
-        assertSame(response, cache.get(projectId2, ip1, databasePath1));
+        assertSame(response, cache.get(projectId2, ip1, databasePath1.toString()));
         // ...and for databasePath2:
-        assertSame(response, cache.get(projectId1, ip1, databasePath2));
+        assertSame(response, cache.get(projectId1, ip1, databasePath2.toString()));
     }
 }

docs/reference/query-languages/esql/kibana/definition/functions/categorize.json

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.otel;
 
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -19,6 +20,10 @@ public class NormalizeForStreamPlugin extends Plugin implements IngestPlugin {
 
     @Override
     public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
-        return Map.of(NormalizeForStreamProcessor.TYPE, new NormalizeForStreamProcessor.Factory());
+        if (DataStream.LOGS_STREAM_FEATURE_FLAG) {
+            return Map.of(NormalizeForStreamProcessor.TYPE, new NormalizeForStreamProcessor.Factory());
+        } else {
+            return Map.of();
+        }
     }
 }

modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpCacheTests.java

@@ -13,6 +13,7 @@
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.FeatureFlag;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
 import org.junit.ClassRule;
@@ -24,7 +25,10 @@ public IngestOtelClientYamlTestSuiteIT(@Name("yaml") ClientYamlTestCandidate tes
     }
 
     @ClassRule
-    public static ElasticsearchCluster cluster = ElasticsearchCluster.local().module("ingest-otel").build();
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .module("ingest-otel")
+        .feature(FeatureFlag.LOGS_STREAM)
+        .build();
 
     @Override
     protected String getTestRestCluster() {

modules/ingest-otel/src/main/java/org/elasticsearch/ingest/otel/NormalizeForStreamPlugin.java

@@ -18,6 +18,7 @@
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
 
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.blobstore.OptionalBytesReference;
 import org.elasticsearch.common.bytes.BytesArray;
@@ -147,7 +148,7 @@ public long absoluteTimeInMillis() {
         // construct our own repo instance so we can inject a threadpool that allows to control the passage of time
         try (
             var repository = new S3Repository(
-                randomProjectIdOrDefault(),
+                ProjectId.DEFAULT,
                 node().injector().getInstance(RepositoriesService.class).repository(TEST_REPO_NAME).getMetadata(),
                 xContentRegistry(),
                 node().injector().getInstance(PluginsService.class).filterPlugins(S3RepositoryPlugin.class).findFirst().get().getService(),