fix: coverage

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

Author: mvallim

amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java

@@ -23,9 +23,11 @@
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
@@ -37,6 +39,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
@@ -46,12 +49,14 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.Mock.Strictness;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import com.amazon.sns.messaging.lib.concurrent.RingBufferBlockingQueue;
 import com.amazon.sns.messaging.lib.core.RequestEntryInternalFactory.RequestEntryInternal;
+import com.amazon.sns.messaging.lib.exception.PoisonRequestEntryException;
 import com.amazon.sns.messaging.lib.helpers.TryConsumer;
 import com.amazon.sns.messaging.lib.model.RequestEntry;
 import com.amazon.sns.messaging.lib.model.ResponseFailEntry;
@@ -112,6 +117,42 @@ void testConstructorInitializesTopicRequests() {
     assertThat(topicRequests.isEmpty(), is(true));
   }
 
+  @Test
+  void testConstructorThrowsNpeWhenTopicPropertyIsNull() {
+    final NullPointerException thrown = assertThrows(NullPointerException.class, () ->
+      new TestableAmazonSnsConsumer(amazonSnsClient, null, objectMapper, pendingRequests, topicRequests, executorService, publishDecorator)
+    );
+
+    assertThat(thrown.getMessage(), containsString("topicProperty cannot be null"));
+  }
+
+  @Test
+  void testConstructorThrowsNpeWhenAmazonSnsClientIsNull() {
+    final NullPointerException thrown = assertThrows(NullPointerException.class, () ->
+      new TestableAmazonSnsConsumer(null, topicProperty, objectMapper, pendingRequests, topicRequests, executorService, publishDecorator)
+    );
+
+    assertThat(thrown.getMessage(), containsString("amazonSnsClient cannot be null"));
+  }
+
+  @Test
+  void testConstructorThrowsNpeWhenObjectMapperIsNull() {
+    final NullPointerException thrown = assertThrows(NullPointerException.class, () ->
+      new TestableAmazonSnsConsumer(amazonSnsClient, topicProperty, null, pendingRequests, topicRequests, executorService, publishDecorator)
+    );
+
+    assertThat(thrown.getMessage(), containsString("objectMapper cannot be null"));
+  }
+
+  @Test
+  void testConstructorThrowsNpeWhenExecutorServiceIsNull() {
+    final NullPointerException thrown = assertThrows(NullPointerException.class, () ->
+      new TestableAmazonSnsConsumer(amazonSnsClient, topicProperty, objectMapper, pendingRequests, topicRequests, null, publishDecorator)
+    );
+
+    assertThat(thrown.getMessage(), containsString("executorService cannot be null"));
+  }
+
   @Test
   void testAwaitReturnCompletableFutureWhenQueuesAreEmpty() throws Exception {
     context(consumer -> {
@@ -251,6 +292,22 @@ void testRunRecordsCorrectExceptionOnPublishFailure() throws Exception {
     });
   }
 
+  @Test
+  void testRunHandlesRejectedExecutionExceptionFromExecutorWithoutCrashing() throws Exception {
+    when(topicProperty.isFifo()).thenReturn(false);
+    doThrow(new RejectedExecutionException("executor full")).when(executorService).execute(any(Runnable.class));
+
+    context(consumer -> {
+      topicRequests.put(buildRequestEntry("rejected-message"));
+
+      await()
+        .untilAsserted(() -> {
+          assertThat(consumer.getHandleErrorCallCount(), greaterThanOrEqualTo(1));
+          assertThat(consumer.getLastError(), instanceOf(RejectedExecutionException.class));
+        });
+    });
+  }
+
   @Test
   void testRunDoesNotPublishWhenQueueIsEmpty() throws Exception {
     context(consumer -> {
@@ -494,6 +551,38 @@ void testCanAddPayloadSplitsBatchWhenCumulativeSizeExceedsThreshold() throws Exc
     });
   }
 
+  @Test
+  void testPoisonRequestEntryRemovesFromPendingRequestsAndFailsListenableFuture() throws Exception {
+    when(topicProperty.isFifo()).thenReturn(true);
+
+    final String poisonId = "poison-id";
+    final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSnsConsumer.batchSizeBytesThreshold() + 100);
+    final RequestEntry<String> poisonEntry = RequestEntry.<String>builder()
+      .withId(poisonId)
+      .withValue(oversizedPayload)
+      .build();
+
+    pendingRequests.put(poisonId, listenableFutureImpl);
+
+    context(consumer -> {
+      topicRequests.put(poisonEntry);
+
+      await()
+        .untilAsserted(() -> {
+          assertThat(pendingRequests.containsKey(poisonId), is(false));
+
+          final ArgumentCaptor<ResponseFailEntry> captor = ArgumentCaptor.forClass(ResponseFailEntry.class);
+          verify(listenableFutureImpl, atLeastOnce()).fail(captor.capture());
+
+          final ResponseFailEntry failEntry = captor.getValue();
+          assertThat(failEntry.getId(), is(poisonId));
+          assertThat(failEntry.getCode(), is("000"));
+          assertThat(failEntry.getSenderFault(), is(true));
+          assertThat(failEntry.getThrowable(), instanceOf(PoisonRequestEntryException.class));
+        });
+    });
+  }
+
   @Test
   void testCanAddPayloadDoesNotPublishEmptyBatchWhenAllEntriesExceedThreshold() throws Exception {
     when(topicProperty.isFifo()).thenReturn(true);