Merge branch 'main' into flush_with_sorting

Author: mgodwan

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowNativeAllocator.java

@@ -179,6 +179,14 @@ public void close() {
             }
         });
         pools.clear();
+        // Close any remaining child allocators (e.g., ad-hoc children created via ArrowAllocatorService)
+        for (BufferAllocator child : new ArrayList<>(root.getChildAllocators())) {
+            try {
+                child.close();
+            } catch (Exception e) {
+                // best-effort — log but don't block shutdown
+            }
+        }
         root.close();
         INSTANCE = null;
     }

plugins/cache-ehcache/src/test/java/org/opensearch/cache/store/disk/EhCacheDiskCacheTests.java

@@ -10,6 +10,7 @@
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 
+import org.apache.lucene.tests.util.LuceneTestCase.AwaitsFix;
 import org.opensearch.cache.EhcacheDiskCacheSettings;
 import org.opensearch.common.Randomness;
 import org.opensearch.common.cache.CacheType;
@@ -354,6 +355,7 @@ public void testEvictions() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Flaky test - fix before enabling")
     public void testComputeIfAbsentConcurrently() throws Exception {
         Settings settings = Settings.builder().build();
         MockRemovalListener<String, String> removalListener = new MockRemovalListener<>();
@@ -558,6 +560,7 @@ public String load(ICacheKey<String> key) throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Flaky test - fix before enabling")
     public void testComputeIfAbsentWithNullValueLoading() throws Exception {
         Settings settings = Settings.builder().build();
         MockRemovalListener<String, String> removalListener = new MockRemovalListener<>();

sandbox/libs/dataformat-native/build.gradle

@@ -13,6 +13,7 @@ description = 'Shared native bridge utilities for OpenSearch sandbox plugins'
 dependencies {
     api project(':libs:opensearch-core')
     api project(':libs:opensearch-common')
+    compileOnly project(':server')
     implementation "org.apache.logging.log4j:log4j-api:${versions.log4j}"
     compileOnly "org.apache.arrow:arrow-c-data:${versions.arrow}"
 

sandbox/libs/dataformat-native/licenses/log4j-api-2.25.4.jar.sha1

@@ -72,6 +72,7 @@ async-trait = "=0.1.89"
 bytes = "=1.11.1"
 criterion = { version = "=0.5.1", features = ["async_tokio"] }
 tokio-metrics = { version = "=0.5.0", features = ["rt"] }
+proptest = "=1.4.0"
 
 # Internal
 native-bridge-common = { path = "common" }

sandbox/libs/dataformat-native/licenses/log4j-api-LICENSE.txt

@@ -73,6 +73,19 @@ public static SymbolLookup symbolLookup() {
         return Holder.LOOKUP;
     }
 
+    /**
+     * Returns {@code true} if the native library has been (or can be) successfully loaded.
+     * This method does not throw — it returns {@code false} if loading fails.
+     */
+    public static boolean isLoaded() {
+        try {
+            symbolLookup();
+            return true;
+        } catch (Throwable t) {
+            return false;
+        }
+    }
+
     // ---- Error handling ----
 
     /**

sandbox/libs/dataformat-native/licenses/log4j-api-NOTICE.txt

@@ -0,0 +1,60 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.nativebridge.spi;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.plugin.stats.AnalyticsBackendNativeMemoryStats;
+
+import java.lang.foreign.FunctionDescriptor;
+import java.lang.foreign.Linker;
+import java.lang.foreign.SymbolLookup;
+import java.lang.foreign.ValueLayout;
+import java.lang.invoke.MethodHandle;
+
+/**
+ * Stateless FFM fetcher for jemalloc memory metrics.
+ * No caching, no server dependencies — just raw downcalls.
+ */
+public class NativeMemoryFetcher {
+
+    private static final Logger logger = LogManager.getLogger(NativeMemoryFetcher.class);
+
+    private static final MethodHandle ALLOCATED;
+    private static final MethodHandle RESIDENT;
+
+    static {
+        SymbolLookup lookup = NativeLibraryLoader.symbolLookup();
+        Linker linker = Linker.nativeLinker();
+        FunctionDescriptor desc = FunctionDescriptor.of(ValueLayout.JAVA_LONG);
+        ALLOCATED = linker.downcallHandle(lookup.find("native_jemalloc_allocated_bytes").orElseThrow(), desc);
+        RESIDENT = linker.downcallHandle(lookup.find("native_jemalloc_resident_bytes").orElseThrow(), desc);
+    }
+
+    private NativeMemoryFetcher() {}
+
+    /**
+     * Performs FFM downcalls and returns a fresh AnalyticsBackendNativeMemoryStats snapshot.
+     * Returns AnalyticsBackendNativeMemoryStats(-1, -1) on error or negative values.
+     */
+    public static AnalyticsBackendNativeMemoryStats fetch() {
+        try {
+            long allocated = (long) ALLOCATED.invokeExact();
+            long resident = (long) RESIDENT.invokeExact();
+            if (allocated < 0 || resident < 0) {
+                logger.warn("Native memory stats returned error: allocated={}, resident={}", allocated, resident);
+                return new AnalyticsBackendNativeMemoryStats(-1, -1);
+            }
+            return new AnalyticsBackendNativeMemoryStats(allocated, resident);
+        } catch (Throwable t) {
+            logger.warn("Error fetching native memory stats", t);
+            return new AnalyticsBackendNativeMemoryStats(-1, -1);
+        }
+    }
+}