test(integration): improve tests

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/concurrent/AmazonSnsThreadPoolExecutor.java

@@ -16,96 +16,15 @@
 
 package com.amazon.sns.messaging.lib.concurrent;
 
-import java.util.Objects;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
-/**
- * A custom {@link ThreadPoolExecutor} that tracks active, failed, and succeeded task counts.
- * Uses a blocking submission policy to handle rejection and a synchronous queue for task handoff.
- */
+// @formatter:off
 public class AmazonSnsThreadPoolExecutor extends ThreadPoolExecutor {
 
-  private final AtomicInteger activeTaskCount = new AtomicInteger();
-
-  private final AtomicInteger failedTaskCount = new AtomicInteger();
-
-  private final AtomicInteger succeededTaskCount = new AtomicInteger();
-
-  /**
-   * Creates a new executor with the specified maximum pool size.
-   *
-   * @param maximumPoolSize the maximum number of threads to allow in the pool
-   */
   public AmazonSnsThreadPoolExecutor(final int maximumPoolSize) {
     super(0, maximumPoolSize, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), ThreadFactoryProvider.getThreadFactory(), new BlockingSubmissionPolicy(30000));
   }
 
-  /**
-   * Returns the number of currently active tasks.
-   *
-   * @return the active task count
-   */
-  public int getActiveTaskCount() {
-    return activeTaskCount.get();
-  }
-
-  /**
-   * Returns the number of tasks that have failed.
-   *
-   * @return the failed task count
-   */
-  public int getFailedTaskCount() {
-    return failedTaskCount.get();
-  }
-
-  /**
-   * Returns the number of tasks that have completed successfully.
-   *
-   * @return the succeeded task count
-   */
-  public int getSucceededTaskCount() {
-    return succeededTaskCount.get();
-  }
-
-  /**
-   * Returns the current size of the task queue.
-   *
-   * @return the queue size
-   */
-  public int getQueueSize() {
-    return getQueue().size();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  protected void beforeExecute(final Thread thread, final Runnable runnable) {
-    try {
-      super.beforeExecute(thread, runnable);
-    } finally {
-      activeTaskCount.incrementAndGet();
-    }
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  protected void afterExecute(final Runnable runnable, final Throwable throwable) {
-    try {
-      super.afterExecute(runnable, throwable);
-    } finally {
-      if (Objects.nonNull(throwable)) {
-        failedTaskCount.incrementAndGet();
-      } else {
-        succeededTaskCount.incrementAndGet();
-      }
-      activeTaskCount.decrementAndGet();
-    }
-  }
-
 }

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/ExecutorsProvider.java

@@ -16,14 +16,9 @@
 
 package com.amazon.sns.messaging.lib.core;
 
-import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicReference;
 
 import com.amazon.sns.messaging.lib.model.RequestEntry;
 import com.amazon.sns.messaging.lib.model.ResponseFailEntry;
@@ -44,14 +39,12 @@
 @RequiredArgsConstructor(access = AccessLevel.PROTECTED)
 abstract class AbstractAmazonSnsProducer<E> {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSnsProducer.class);
+  private final AtomicReference<State> state = new AtomicReference<>(State.RUNNIG);
 
   private final ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests;
 
   private final BlockingQueue<RequestEntry<E>> topicRequests;
 
-  private final ExecutorService executorService;
-
   /**
    * Sends a request entry by enqueuing it for batch processing.
    *
@@ -60,25 +53,27 @@ abstract class AbstractAmazonSnsProducer<E> {
    */
   @SneakyThrows
   public ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> send(final RequestEntry<E> requestEntry) {
-    return enqueueRequest(requestEntry);
-  }
+    if (State.RUNNIG.equals(state.get())) {
+      return enqueueRequest(requestEntry);
+    } else {
+      final ListenableFutureImpl listenableFutureImpl = new ListenableFutureImpl();
 
-  /**
-   * Shuts down the producer's executor service gracefully, waiting up to 60 seconds
-   * for termination.
-   */
-  @SneakyThrows
-  public void shutdown() {
-    LOGGER.warn("Shutdown producer {}", getClass().getSimpleName());
+      listenableFutureImpl.fail(ResponseFailEntry.builder()
+        .withCode("000")
+        .withId(requestEntry.getId())
+        .withMessage(String.format("Producer is currently in %s mode; no further messages will be accepted.", state.get().name()))
+        .withSenderFault(true)
+        .build()
+      );
 
-    executorService.shutdown();
-    if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
-      LOGGER.warn("Executor service did not terminate in the specified time.");
-      final List<Runnable> droppedTasks = executorService.shutdownNow();
-      LOGGER.warn("Executor service was abruptly shut down. {} tasks will not be executed.", droppedTasks.size());
+      return listenableFutureImpl;
     }
   }
 
+  public void shutdown() {
+    state.compareAndSet(State.RUNNIG, State.SHUTDOWN);
+  }
+
   /**
    * Creates a {@link ListenableFuture} for the request, registers it in the pending map,
    * and enqueues the request for batch processing.
@@ -94,5 +89,9 @@ private ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> enqueueRequest
     return trackPendingRequest;
   }
 
+  enum State {
+    RUNNIG, SHUTDOWN
+  }
+
 }
 // @formatter:on

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

@@ -16,15 +16,29 @@
 
 package com.amazon.sns.messaging.lib.core;
 
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 import com.amazon.sns.messaging.lib.concurrent.AmazonSnsThreadPoolExecutor;
+import com.amazon.sns.messaging.lib.concurrent.RingBufferBlockingQueue;
+import com.amazon.sns.messaging.lib.metrics.BlockingQueueMetricsDecorator;
+import com.amazon.sns.messaging.lib.metrics.ExecutorServiceMetricsDecorator;
 import com.amazon.sns.messaging.lib.model.RequestEntry;
 import com.amazon.sns.messaging.lib.model.ResponseFailEntry;
 import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry;
 import com.amazon.sns.messaging.lib.model.TopicProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import lombok.AccessLevel;
+import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
 // @formatter:off
@@ -79,8 +93,80 @@ public CompletableFuture<Void> await() {
    * @param topicProperty the topic configuration
    * @return a configured thread pool executor
    */
-  protected static AmazonSnsThreadPoolExecutor getAmazonSnsThreadPoolExecutor(final TopicProperty topicProperty) {
-    return topicProperty.isFifo() ? new AmazonSnsThreadPoolExecutor(1) : new AmazonSnsThreadPoolExecutor(topicProperty.getMaximumPoolSize());
+  protected static ExecutorService getExecutorService(final TopicProperty topicProperty, final MeterRegistry meterRegistry) {
+    return topicProperty.isFifo()
+      ? new ExecutorServiceMetricsDecorator(
+          new AmazonSnsThreadPoolExecutor(1),
+          meterRegistry,
+          topicProperty.getTopicArn()
+        )
+      : new ExecutorServiceMetricsDecorator(
+          new AmazonSnsThreadPoolExecutor(topicProperty.getMaximumPoolSize()),
+          meterRegistry,
+          topicProperty.getTopicArn()
+        );
+  }
+
+  @Getter
+  public static final class Builder<C, R, O, E, T extends AbstractAmazonSnsTemplate<C, R, O, E>> {
+
+    private final C amazonSnsClient;
+
+    private final TopicProperty topicProperty;
+
+    private ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests = new ConcurrentHashMap<>();
+
+    private BlockingQueue<RequestEntry<E>> topicRequests;
+
+    private ObjectMapper objectMapper = new ObjectMapper();
+
+    private UnaryOperator<R> publishDecorator = UnaryOperator.identity();
+
+    private MeterRegistry meterRegistry = new SimpleMeterRegistry();
+
+    private final Function<Builder<C, R, O, E, T>, T> constructor;
+
+    Builder(final Function<Builder<C, R, O, E, T>, T> constructor, final C amazonSnsClient, final TopicProperty topicProperty) {
+      this.amazonSnsClient = Objects.requireNonNull(amazonSnsClient, "amazonSnsClient");
+      this.topicProperty = Objects.requireNonNull(topicProperty, "topicProperty");
+      this.constructor = Objects.requireNonNull(constructor, "constructor");
+    }
+
+    public Builder<C, R, O, E, T> pendingRequests(final ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests) {
+      this.pendingRequests = Objects.requireNonNull(pendingRequests, "pendingRequests");
+      return this;
+    }
+
+    public Builder<C, R, O, E, T> topicRequests(final BlockingQueue<RequestEntry<E>> topicRequests) {
+      this.topicRequests = Objects.requireNonNull(topicRequests, "topicRequests");
+      return this;
+    }
+
+    public Builder<C, R, O, E, T> objectMapper(final ObjectMapper objectMapper) {
+      this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper");
+      return this;
+    }
+
+    public Builder<C, R, O, E, T> publishDecorator(final UnaryOperator<R> publishDecorator) {
+      this.publishDecorator = Objects.requireNonNull(publishDecorator, "publishDecorator");
+      return this;
+    }
+
+    public Builder<C, R, O, E, T> meterRegistry(final MeterRegistry meterRegistry) {
+      this.meterRegistry = Objects.requireNonNull(meterRegistry, "meterRegistry");
+      return this;
+    }
+
+    public T build() {
+      if (Objects.isNull(topicRequests)) {
+        topicRequests = new RingBufferBlockingQueue<>(topicProperty.getMaximumPoolSize() * topicProperty.getMaxBatchSize());
+      }
+
+      topicRequests = new BlockingQueueMetricsDecorator<>(topicRequests, meterRegistry, topicProperty.getTopicArn());
+
+      return constructor.apply(this);
+    }
+
   }
 
 }