feat: add instrumentation MBean

Signed-off-by: Marcos Tischer Vallim <tischer@gmail.com>

Author: mvallim

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsThreadPoolExecutor.java

@@ -22,50 +22,56 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-public class AmazonSnsThreadPoolExecutor extends ThreadPoolExecutor {
+import com.amazon.sns.messaging.lib.concurrent.ThreadFactoryProvider;
 
+public class AmazonSnsThreadPoolExecutor extends ThreadPoolExecutor {
+  
   private final AtomicInteger activeTaskCount = new AtomicInteger();
-
+  
   private final AtomicInteger failedTaskCount = new AtomicInteger();
-
+  
   private final AtomicInteger succeededTaskCount = new AtomicInteger();
-
+  
   public AmazonSnsThreadPoolExecutor(final int maximumPoolSize) {
-    super(0, maximumPoolSize, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), new BlockingSubmissionPolicy(30000));
+    super(0, maximumPoolSize, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), ThreadFactoryProvider.getThreadFactory(), new BlockingSubmissionPolicy(30000));
   }
-
+  
   public int getActiveTaskCount() {
-    return activeTaskCount.get();
+    return this.activeTaskCount.get();
   }
-
+  
   public int getFailedTaskCount() {
-    return failedTaskCount.get();
+    return this.failedTaskCount.get();
   }
-
+  
   public int getSucceededTaskCount() {
-    return succeededTaskCount.get();
+    return this.succeededTaskCount.get();
   }
-
+  
+  public int getQueueSize() {
+    return getQueue().size();
+  }
+  
   @Override
   protected void beforeExecute(final Thread thread, final Runnable runnable) {
     try {
       super.beforeExecute(thread, runnable);
     } finally {
-      activeTaskCount.incrementAndGet();
+      this.activeTaskCount.incrementAndGet();
     }
   }
-
+  
   @Override
   protected void afterExecute(final Runnable runnable, final Throwable throwable) {
     try {
       super.afterExecute(runnable, throwable);
     } finally {
       if (Objects.nonNull(throwable)) {
-        failedTaskCount.incrementAndGet();
+        this.failedTaskCount.incrementAndGet();
       } else {
-        succeededTaskCount.incrementAndGet();
+        this.succeededTaskCount.incrementAndGet();
       }
-      activeTaskCount.decrementAndGet();
+      this.activeTaskCount.decrementAndGet();
     }
   }
 

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/instrument/AmazonSnsThreadPoolExecutorJmx.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.amazon.sns.messaging.lib.instrument;
+
+import com.amazon.sns.messaging.lib.core.AmazonSnsThreadPoolExecutor;
+
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class AmazonSnsThreadPoolExecutorJmx implements AmazonSnsThreadPoolExecutorJmxMBean {
+
+  private final AmazonSnsThreadPoolExecutor amazonSnsThreadPoolExecutor;
+
+  @Override
+  public int getActiveTaskCount() {
+    return this.amazonSnsThreadPoolExecutor.getActiveTaskCount();
+  }
+
+  @Override
+  public int getFailedTaskCount() {
+    return this.amazonSnsThreadPoolExecutor.getFailedTaskCount();
+  }
+
+  @Override
+  public int getSucceededTaskCount() {
+    return this.amazonSnsThreadPoolExecutor.getSucceededTaskCount();
+  }
+
+  @Override
+  public int getQueueSize() {
+    return this.amazonSnsThreadPoolExecutor.getQueueSize();
+  }
+
+  @Override
+  public int getPoolSize() {
+    return this.amazonSnsThreadPoolExecutor.getPoolSize();
+  }
+  
+}

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/instrument/AmazonSnsThreadPoolExecutorJmxMBean.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.amazon.sns.messaging.lib.instrument;
+
+public interface AmazonSnsThreadPoolExecutorJmxMBean {
+  
+  int getActiveTaskCount();
+
+  int getFailedTaskCount();
+
+  int getSucceededTaskCount();
+
+  int getQueueSize();
+
+  int getPoolSize();
+
+}

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/instrument/MBeanRegistrar.java

@@ -0,0 +1,37 @@
+package com.amazon.sns.messaging.lib.instrument;
+
+import java.lang.management.ManagementFactory;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.SneakyThrows;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class MBeanRegistrar {
+
+  private static final MBeanServer BEAN_SERVER = ManagementFactory.getPlatformMBeanServer();
+
+  @SneakyThrows
+  public static void registerMBean(final Object mbean, final String name) {
+    final ObjectName objectName = new ObjectName(name);
+
+    if (!MBeanRegistrar.BEAN_SERVER.isRegistered(objectName)) {
+      MBeanRegistrar.BEAN_SERVER.registerMBean(mbean, objectName);
+    }
+    
+  }
+
+  @SneakyThrows
+  public static void unregisterMBean(final String name) {
+    final ObjectName objectName = new ObjectName(name);
+    
+    if (MBeanRegistrar.BEAN_SERVER.isRegistered(objectName)) {
+      MBeanRegistrar.BEAN_SERVER.unregisterMBean(objectName);
+    }
+    
+  }
+  
+}

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/instrument/RingBufferBlockingQueueJmx.java

@@ -0,0 +1,42 @@
+package com.amazon.sns.messaging.lib.instrument;
+
+import com.amazon.sns.messaging.lib.concurrent.RingBufferBlockingQueue;
+
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class RingBufferBlockingQueueJmx implements RingBufferBlockingQueueJmxMBean {
+
+  private final RingBufferBlockingQueue<?> ringBufferBlockingQueue;
+
+  @Override
+  public int capacity() {
+    return this.ringBufferBlockingQueue.capacity();
+  }
+
+  @Override
+  public int size() {
+    return this.ringBufferBlockingQueue.size();
+  }
+
+  @Override
+  public boolean isEmpty() {
+    return this.ringBufferBlockingQueue.isEmpty();
+  }
+
+  @Override
+  public boolean isFull() {
+    return this.ringBufferBlockingQueue.isFull();
+  }
+
+  @Override
+  public long writeSequence() {
+    return this.ringBufferBlockingQueue.writeSequence();
+  }
+
+  @Override
+  public long readSequence() {
+    return this.ringBufferBlockingQueue.readSequence();
+  }
+
+}

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/instrument/RingBufferBlockingQueueJmxMBean.java

@@ -0,0 +1,17 @@
+package com.amazon.sns.messaging.lib.instrument;
+
+public interface RingBufferBlockingQueueJmxMBean {
+
+  int capacity();
+
+  int size();
+
+  boolean isEmpty();
+
+  boolean isFull();
+
+  long writeSequence();
+
+  long readSequence();
+
+}