@Test(expected = Exception.class)
public void testConsumeInternalsFailure() {

  urlSetup("testQueue", "testUrl");
  receiveSetup();
  // dummy reset visibility
  ChangeMessageVisibilityRequest changeVisibilityReq =
      ChangeMessageVisibilityRequest.builder()
          .queueUrl("testUrl")
          .receiptHandle("testHandle")
          .visibilityTimeout(0)
          .build();
  CompletableFuture<ChangeMessageVisibilityResponse> visibilityResponse =
      CompletableFuture.completedFuture(ChangeMessageVisibilityResponse.builder().build());
  when(awsAsyncSqsClient.changeMessageVisibility(changeVisibilityReq))
      .thenReturn(visibilityResponse);
  ArgumentCaptor<ChangeMessageVisibilityRequest> visibilityReq =
      ArgumentCaptor.forClass(ChangeMessageVisibilityRequest.class);
  CompletableFuture<Void> future = new CompletableFuture<>();
  future.completeExceptionally(new Exception("async consumer completing exceptionally"));
  AsyncConsumer<String> function = (str, threadpool) -> future;
  CompletableFuture<Boolean> result = asyncClient.consume(function, "testQueue");
  // testing setMessageVisible
  verify(awsAsyncSqsClient).changeMessageVisibility(visibilityReq.capture());
  assertEquals(visibilityReq.getValue().queueUrl(), "testUrl");
  assertEquals(visibilityReq.getValue().receiptHandle(), "testHandle");
  assertEquals(visibilityReq.getValue().visibilityTimeout(), new Integer(0));
  verify(awsAsyncSqsClient, times(1))
      .changeMessageVisibility(any(ChangeMessageVisibilityRequest.class));
  // testing final result
  result.join();
}