Limit memory consumption in test on overflow (#17373)

Updates only test code to be able to run a test that consumes big memory if:
- the physical memory is bigger than the requested Java heap
- JDK version is greater than or equal to 21.

The reason to limit the JDK version is that on 16GB machine the G1GC is more efficient than the one on previous JDKs and so let complete the test with 10GB heap, while in JDK 17 it consistently fails with OOM error.

logstash-core/build.gradle

@@ -124,7 +124,8 @@ tasks.register("javaTests", Test) {
     exclude '/org/logstash/plugins/factory/PluginFactoryExtTest.class'
     exclude '/org/logstash/execution/ObservedExecutionTest.class'
 
-    maxHeapSize = "12g"
+    // 10GB is needed by the BufferedTokenizerExtWithSizeLimitTest.givenTooLongInputExtractDoesntOverflow test
+    maxHeapSize = "10g"
 
     jacoco {
         enabled = true

logstash-core/src/main/java/org/logstash/util/JavaVersion.java

@@ -31,6 +31,7 @@ public class JavaVersion implements Comparable<JavaVersion> {
 
     public static final JavaVersion CURRENT = parse(System.getProperty("java.specification.version"));
     public static final JavaVersion JAVA_17 = parse("17");
+    public static final JavaVersion JAVA_21 = parse("21");
 
     private final List<Integer> version;
 

logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithSizeLimitTest.java

@@ -19,6 +19,7 @@
 
 package org.logstash.common;
 
+import org.hamcrest.Matchers;
 import org.jruby.RubyArray;
 import org.jruby.RubyString;
 import org.jruby.runtime.ThreadContext;
@@ -27,12 +28,21 @@ import org.junit.Before;
 import org.junit.Test;
 import org.logstash.RubyTestBase;
 import org.logstash.RubyUtil;
+import org.logstash.util.JavaVersion;
 
+import javax.management.Attribute;
+import javax.management.InstanceNotFoundException;
+import javax.management.ReflectionException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
 import java.util.List;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assume.assumeThat;
+import static org.junit.Assume.assumeTrue;
 import static org.logstash.RubyUtil.RUBY;
 
 @SuppressWarnings("unchecked")
@@ -115,27 +125,67 @@ public final class BufferedTokenizerExtWithSizeLimitTest extends RubyTestBase {
     }
 
     @Test
-    public void givenMaliciousInputExtractDoesntOverflow() {
+    public void givenTooLongInputExtractDoesntOverflow() {
+        // This test has proven to go OOM on JDK 11 and JDK 17, also if physical memory is 16GB.
+        // JDK 21 successfully executes the test due to internal changes or efficiency of G1GC (the default GC).
+        // Tested also others GC on JDK 11 without any success:
+        // - ZGC
+        // - Parallel GC
+        // - CMS
+        // remove this code when the minimal JDK version for Logstash is JDK 21 or greater.
+
+        assumeThat("Expect at least JDK 21", JavaVersion.CURRENT, Matchers.greaterThanOrEqualTo(JavaVersion.JAVA_21));
+        long expectedNeedHeapMemory = 10L * GB;
+        // To understand the motivation of 10GB heap please read https://github.com/elastic/logstash/pull/17373#issuecomment-2750378212
+        assumeTrue("Skip the test because VM hasn't enough physical memory", hasEnoughPhysicalMemory(expectedNeedHeapMemory));
+
         assertEquals("Xmx must equals to what's defined in the Gradle's javaTests task",
-                12L * GB, Runtime.getRuntime().maxMemory());
+                expectedNeedHeapMemory, Runtime.getRuntime().maxMemory());
 
         // re-init the tokenizer with big sizeLimit
         initSUTWithSizeLimit((int) (2L * GB) - 3);
         // Integer.MAX_VALUE is 2 * GB
-        String bigFirstPiece = generateString("a", Integer.MAX_VALUE - 1024);
-        sut.extract(context, RubyUtil.RUBY.newString(bigFirstPiece));
+        RubyString bigFirstPiece = generateString("a", Integer.MAX_VALUE - 1024);
+        sut.extract(context, bigFirstPiece);
 
         // add another small fragment to trigger int overflow
         // sizeLimit is (2^32-1)-3 first segment length is (2^32-1) - 1024 second is 1024 +2
-        // so the combined length of first and second is > sizeLimit and should throw an expection
+        // so the combined length of first and second is > sizeLimit and should throw an exception
         // but because of overflow it's negative and happens to be < sizeLimit
         Exception thrownException = assertThrows(IllegalStateException.class, () -> {
-            sut.extract(context, RubyUtil.RUBY.newString(generateString("a", 1024 + 2)));
+            sut.extract(context, generateString("a", 1024 + 2));
         });
         assertThat(thrownException.getMessage(), containsString("input buffer full"));
     }
 
-    private String generateString(String fill, int size) {
-        return fill.repeat(size);
+    private RubyString generateString(String fill, int size) {
+        return RubyUtil.RUBY.newString(fill.repeat(size));
+    }
+
+    private boolean hasEnoughPhysicalMemory(long requiredPhysicalMemory) {
+        long physicalMemory;
+        try {
+            physicalMemory = readPhysicalMemorySize();
+        } catch (InstanceNotFoundException | ReflectionException e) {
+            System.out.println("Can't read attribute JMX OS bean");
+            return false;
+        } catch (IllegalStateException e) {
+            System.out.println(e.getMessage());
+            return false;
+        }
+        System.out.println("Physical memory on the VM is: " + physicalMemory + " bytes");
+        return physicalMemory > requiredPhysicalMemory;
+    }
+
+    private long readPhysicalMemorySize() throws ReflectionException, InstanceNotFoundException {
+        OperatingSystemMXBean op = ManagementFactory.getOperatingSystemMXBean();
+
+        List<Attribute> attributes = ManagementFactory.getPlatformMBeanServer()
+                .getAttributes(op.getObjectName(), new String[]{"TotalPhysicalMemorySize"} ).asList();
+        if (attributes.isEmpty()) {
+            throw new IllegalStateException("Attribute TotalPhysicalMemorySize is not available from JMX OS bean");
+        }
+        Attribute a = attributes.get(0);
+        return (long) (Long) a.getValue();
     }
 }