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