### Question:
ShardConsumer { @VisibleForTesting synchronized CompletableFuture<Boolean> initializeComplete() { if (taskOutcome != null) { updateState(taskOutcome); } if (currentState.state() == ConsumerStates.ShardConsumerState.PROCESSING) { return CompletableFuture.completedFuture(true); } return CompletableFuture.supplyAsync(() -> { if (isShutdownRequested()) { throw new IllegalStateException("Shutdown requested while initializing"); } executeTask(null); if (isShutdownRequested()) { throw new IllegalStateException("Shutdown requested while initializing"); } return false; }, executorService); } @Deprecated  ShardConsumer(RecordsPublisher recordsPublisher, ExecutorService executorService, ShardInfo shardInfo,
            Optional<Long> logWarningForTaskAfterMillis, ShardConsumerArgument shardConsumerArgument,
            TaskExecutionListener taskExecutionListener);  ShardConsumer(RecordsPublisher recordsPublisher, ExecutorService executorService, ShardInfo shardInfo,
            Optional<Long> logWarningForTaskAfterMillis, ShardConsumerArgument shardConsumerArgument,
            TaskExecutionListener taskExecutionListener, int readTimeoutsToIgnoreBeforeWarning); @Deprecated  ShardConsumer(RecordsPublisher recordsPublisher, ExecutorService executorService, ShardInfo shardInfo,
            Optional<Long> logWarningForTaskAfterMillis, ShardConsumerArgument shardConsumerArgument,
            ConsumerState initialState, Function<ConsumerTask, ConsumerTask> taskMetricsDecorator, int bufferSize,
            TaskExecutionListener taskExecutionListener);  ShardConsumer(RecordsPublisher recordsPublisher, ExecutorService executorService, ShardInfo shardInfo,
            Optional<Long> logWarningForTaskAfterMillis, ShardConsumerArgument shardConsumerArgument,
            ConsumerState initialState, Function<ConsumerTask, ConsumerTask> taskMetricsDecorator, int bufferSize,
            TaskExecutionListener taskExecutionListener, int readTimeoutsToIgnoreBeforeWarning); void executeLifecycle(); void gracefulShutdown(ShutdownNotification shutdownNotification); boolean leaseLost(); boolean isShutdown(); @VisibleForTesting boolean isShutdownRequested(); static final int MAX_TIME_BETWEEN_REQUEST_RESPONSE; }### Answer:
@SuppressWarnings("unchecked") @Test(expected = RejectedExecutionException.class) public final void testInitializationStateUponSubmissionFailure() throws Exception { ExecutorService failingService = mock(ExecutorService.class); ShardConsumer consumer = new ShardConsumer(recordsPublisher, failingService, shardInfo, logWarningForTaskAfterMillis, shardConsumerArgument, initialState, t -> t, 1, taskExecutionListener, 0); doThrow(new RejectedExecutionException()).when(failingService).execute(any()); boolean initComplete; do { initComplete = consumer.initializeComplete().get(); } while (!initComplete); verifyZeroInteractions(taskExecutionListener); }
@Test public void testErrorThrowableInInitialization() throws Exception { ShardConsumer consumer = new ShardConsumer(recordsPublisher, executorService, shardInfo, logWarningForTaskAfterMillis, shardConsumerArgument, initialState, t -> t, 1, taskExecutionListener, 0); when(initialState.createTask(any(), any(), any())).thenReturn(initializeTask); when(initialState.taskType()).thenReturn(TaskType.INITIALIZE); when(initializeTask.call()).thenAnswer(i -> { throw new Error("Error"); }); try { consumer.initializeComplete().get(); } catch (ExecutionException ee) { assertThat(ee.getCause(), instanceOf(Error.class)); } verify(taskExecutionListener, times(1)).beforeTaskExecution(initialTaskInput); verifyNoMoreInteractions(taskExecutionListener); }