@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); }
@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); }
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); } }
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); }
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(); }
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; }