Throw a configuration error if Netty reserved direct memory is not an…

…ough (256MB)

lib/logstash/inputs/beats.rb

@@ -250,6 +250,8 @@ def create_server
     server = org.logstash.beats.Server.new(@host, @port, @client_inactivity_timeout, @executor_threads, @protect_direct_memory)
     server.setSslHandlerProvider(new_ssl_handshake_provider(new_ssl_context_builder)) if @ssl_enabled
     server
+  rescue java.lang.IllegalArgumentException => e
+    configuration_error e.message
   end
 
   def run(output_queue)

src/main/java/org/logstash/beats/Server.java

@@ -12,6 +12,7 @@
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.DefaultEventExecutorGroup;
 import io.netty.util.concurrent.EventExecutorGroup;
+import io.netty.util.internal.PlatformDependent;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.logstash.netty.SslHandlerProvider;
@@ -30,16 +31,26 @@ public class Server {
 
     private final int clientInactivityTimeoutSeconds;
 
-//    public Server(String host, int p, int clientInactivityTimeoutSeconds, int threadCount) {
-//        this(host, p, clientInactivityTimeoutSeconds, threadCount, true);
-//    }
-
     public Server(String host, int p, int clientInactivityTimeoutSeconds, int threadCount, boolean protectDirectMemory) {
         this.host = host;
         port = p;
         this.clientInactivityTimeoutSeconds = clientInactivityTimeoutSeconds;
         beatsHeandlerThreadCount = threadCount;
         this.protectDirectMemory = protectDirectMemory;
+
+        validateMinimumDirectMemory();
+    }
+
+    /**
+     * Validate if the configured available direct memory is enough for safe processing, else throws a ConfigurationException
+     * */
+    private void validateMinimumDirectMemory() {
+        long maxDirectMemoryAllocatable = PlatformDependent.maxDirectMemory();
+        if (maxDirectMemoryAllocatable < 256 * 1024 * 1024) {
+            long roundedMegabytes = Math.round((double) maxDirectMemoryAllocatable / 1024 / 1024);
+            throw new IllegalArgumentException("Max direct memory should be at least 256MB but was " + roundedMegabytes + "MB, " +
+                    "please check your MaxDirectMemorySize and io.netty.maxDirectMemory settings");
+        }
     }
 
     public void setSslHandlerProvider(SslHandlerProvider sslHandlerProvider){