properly name netty threads

lib/logstash/inputs/beats.rb

@@ -205,7 +205,7 @@ def register
   end # def register
 
   def create_server
-    server = org.logstash.beats.Server.new(@host, @port, @client_inactivity_timeout, @event_loop_threads, @executor_threads)
+    server = org.logstash.beats.Server.new(@id, @host, @port, @client_inactivity_timeout, @event_loop_threads, @executor_threads)
     server.setSslHandlerProvider(new_ssl_handshake_provider(new_ssl_context_builder)) if @ssl_enabled
     server
   end

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

@@ -17,7 +17,7 @@ static public void main(String[] args) throws Exception {
         // Check for leaks.
         // ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID);
 
-        Server server = new Server("0.0.0.0", DEFAULT_PORT, 15, 0, Runtime.getRuntime().availableProcessors());
+        Server server = new Server("test", "0.0.0.0", DEFAULT_PORT, 15, 0, Runtime.getRuntime().availableProcessors());
 
             if(args.length > 0 && args[0].equals("ssl")) {
             logger.debug("Using SSL");

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

@@ -18,10 +18,14 @@
 import org.apache.logging.log4j.Logger;
 import org.logstash.netty.SslHandlerProvider;
 
+import static org.logstash.beats.util.DaemonThreadFactory.daemonThreadFactory;
+
 public class Server {
     private final static Logger logger = LogManager.getLogger(Server.class);
 
     private final int port;
+
+    private final String id;
     private final String host;
     private final int eventLoopThreadCount;
     private final int executorThreadCount;
@@ -33,7 +37,8 @@ public class Server {
 
     private final int clientInactivityTimeoutSeconds;
 
-    public Server(String host, int port, int clientInactivityTimeoutSeconds, int eventLoopThreadCount, int executorThreadCount) {
+    public Server(String id, String host, int port, int clientInactivityTimeoutSeconds, int eventLoopThreadCount, int executorThreadCount) {
+        this.id = id;
         this.host = host;
         this.port = port;
         this.clientInactivityTimeoutSeconds = clientInactivityTimeoutSeconds;
@@ -54,12 +59,12 @@ public Server listen() throws InterruptedException {
                 logger.error("Could not shut down worker group before starting", e);
             }
         }
-        bossGroup = new NioEventLoopGroup(eventLoopThreadCount); // TODO: add a config to make it adjustable, no need many threads
-        workGroup = new NioEventLoopGroup(eventLoopThreadCount);
+        bossGroup = new NioEventLoopGroup(eventLoopThreadCount, daemonThreadFactory(id + "-bossGroup")); // TODO: add a config to make it adjustable, no need many threads
+        workGroup = new NioEventLoopGroup(eventLoopThreadCount, daemonThreadFactory(id + "-workGroup"));
         try {
             logger.info("Starting server on port: {}", this.port);
 
-            beatsInitializer = new BeatsInitializer(messageListener, clientInactivityTimeoutSeconds, executorThreadCount);
+            beatsInitializer = new BeatsInitializer(id, messageListener, clientInactivityTimeoutSeconds, executorThreadCount);
 
             ServerBootstrap server = new ServerBootstrap();
             server.group(bossGroup, workGroup)
@@ -143,12 +148,14 @@ private class BeatsInitializer extends ChannelInitializer<SocketChannel> {
         private final IMessageListener localMessageListener;
         private final int localClientInactivityTimeoutSeconds;
 
-        BeatsInitializer(IMessageListener messageListener, int clientInactivityTimeoutSeconds, int beatsHandlerThread) {
+        BeatsInitializer(String pluginId, IMessageListener messageListener, int clientInactivityTimeoutSeconds, int beatsHandlerThreadCount) {
             // Keeps a local copy of Server settings, so they can't be modified once it starts listening
             this.localMessageListener = messageListener;
             this.localClientInactivityTimeoutSeconds = clientInactivityTimeoutSeconds;
-            idleExecutorGroup = new DefaultEventExecutorGroup(DEFAULT_IDLESTATEHANDLER_THREAD);
-            beatsHandlerExecutorGroup = new DefaultEventExecutorGroup(beatsHandlerThread);
+            idleExecutorGroup = new DefaultEventExecutorGroup(DEFAULT_IDLESTATEHANDLER_THREAD,
+                    daemonThreadFactory(pluginId + "-idleStateHandler"));
+            beatsHandlerExecutorGroup = new DefaultEventExecutorGroup(beatsHandlerThreadCount,
+                    daemonThreadFactory(pluginId + "-beatsHandler"));
         }
 
         public void initChannel(SocketChannel socket) {