Use internal thread factory

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/ScheduledExecutorServiceWrapper.java

@@ -13,7 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
-import io.micrometer.core.instrument.util.NamedThreadFactory;
+import com.rabbitmq.stream.impl.Utils.NamedThreadFactory;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -212,8 +212,7 @@ class StreamEnvironment implements Environment {
       executorService = scheduledExecutorService;
       this.privateScheduleExecutorService = false;
     }
-    // TODO remove executor wrapper (it's here just for debugging)
-    this.scheduledExecutorService = new ScheduledExecutorServiceWrapper(executorService);
+    this.scheduledExecutorService = executorService;
 
     this.producersCoordinator =
         new ProducersCoordinator(

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -34,6 +34,8 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -482,4 +484,29 @@ static String jsonField(String name, Number value) {
   static String jsonField(String name, String value) {
     return quote(name) + " : " + quote(value);
   }
+
+  static class NamedThreadFactory implements ThreadFactory {
+
+    private final ThreadFactory backingThreaFactory;
+
+    private final String prefix;
+
+    private final AtomicLong count = new AtomicLong(0);
+
+    public NamedThreadFactory(String prefix) {
+      this(Executors.defaultThreadFactory(), prefix);
+    }
+
+    public NamedThreadFactory(ThreadFactory backingThreadFactory, String prefix) {
+      this.backingThreaFactory = backingThreadFactory;
+      this.prefix = prefix;
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+      Thread thread = this.backingThreaFactory.newThread(r);
+      thread.setName(prefix + count.getAndIncrement());
+      return thread;
+    }
+  }
 }