@SuppressWarnings("unchecked") @Test public final void testConsumerTransitionsToProcessingWithIntermittentInitializationFailures() { ExecutorService failingService = spy(executorService); ShardConsumer consumer = new ShardConsumer(recordsPublisher, failingService, shardInfo, logWarningForTaskAfterMillis, shardConsumerArgument, blockedOnParentsState, t -> t, 1, taskExecutionListener, 0); mockSuccessfulUnblockOnParents(); mockSuccessfulInitializeWithFailureTransition(); mockSuccessfulProcessing(null); doAnswer(invocation -> directlyExecuteRunnable(invocation)) .doThrow(new RejectedExecutionException()) .doThrow(new RejectedExecutionException()) .doThrow(new RejectedExecutionException()) .doAnswer(invocation -> directlyExecuteRunnable(invocation)) .when(failingService).execute(any()); int arbitraryExecutionCount = 6; do { try { consumer.executeLifecycle(); } catch (Exception e) { fail("Unexpected exception while executing consumer lifecycle"); } } while (--arbitraryExecutionCount > 0); assertEquals(ShardConsumerState.PROCESSING.consumerState().state(), consumer.currentState().state()); verify(failingService, times(5)).execute(any()); }
public void executeLifecycle() { if (isShutdown()) { return; } if (stateChangeFuture != null && !stateChangeFuture.isDone()) { return; } try { if (isShutdownRequested()) { stateChangeFuture = shutdownComplete(); } else if (needsInitialization) { if (stateChangeFuture != null) { if (stateChangeFuture.get()) { subscribe(); needsInitialization = false; } } stateChangeFuture = initializeComplete(); } } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } catch (RejectedExecutionException e) { taskOutcome = TaskOutcome.FAILURE; } if (ConsumerStates.ShardConsumerState.PROCESSING.equals(currentState.state())) { Throwable t = healthCheck(); if (t instanceof Error) { throw (Error) t; } } }
ShardConsumer { public void executeLifecycle() { if (isShutdown()) { return; } if (stateChangeFuture != null && !stateChangeFuture.isDone()) { return; } try { if (isShutdownRequested()) { stateChangeFuture = shutdownComplete(); } else if (needsInitialization) { if (stateChangeFuture != null) { if (stateChangeFuture.get()) { subscribe(); needsInitialization = false; } } stateChangeFuture = initializeComplete(); } } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } catch (RejectedExecutionException e) { taskOutcome = TaskOutcome.FAILURE; } if (ConsumerStates.ShardConsumerState.PROCESSING.equals(currentState.state())) { Throwable t = healthCheck(); if (t instanceof Error) { throw (Error) t; } } } }
ShardConsumer { public void executeLifecycle() { if (isShutdown()) { return; } if (stateChangeFuture != null && !stateChangeFuture.isDone()) { return; } try { if (isShutdownRequested()) { stateChangeFuture = shutdownComplete(); } else if (needsInitialization) { if (stateChangeFuture != null) { if (stateChangeFuture.get()) { subscribe(); needsInitialization = false; } } stateChangeFuture = initializeComplete(); } } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } catch (RejectedExecutionException e) { taskOutcome = TaskOutcome.FAILURE; } if (ConsumerStates.ShardConsumerState.PROCESSING.equals(currentState.state())) { Throwable t = healthCheck(); if (t instanceof Error) { throw (Error) t; } } } @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 { public void executeLifecycle() { if (isShutdown()) { return; } if (stateChangeFuture != null && !stateChangeFuture.isDone()) { return; } try { if (isShutdownRequested()) { stateChangeFuture = shutdownComplete(); } else if (needsInitialization) { if (stateChangeFuture != null) { if (stateChangeFuture.get()) { subscribe(); needsInitialization = false; } } stateChangeFuture = initializeComplete(); } } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } catch (RejectedExecutionException e) { taskOutcome = TaskOutcome.FAILURE; } if (ConsumerStates.ShardConsumerState.PROCESSING.equals(currentState.state())) { Throwable t = healthCheck(); if (t instanceof Error) { throw (Error) t; } } } @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 { public void executeLifecycle() { if (isShutdown()) { return; } if (stateChangeFuture != null && !stateChangeFuture.isDone()) { return; } try { if (isShutdownRequested()) { stateChangeFuture = shutdownComplete(); } else if (needsInitialization) { if (stateChangeFuture != null) { if (stateChangeFuture.get()) { subscribe(); needsInitialization = false; } } stateChangeFuture = initializeComplete(); } } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } catch (RejectedExecutionException e) { taskOutcome = TaskOutcome.FAILURE; } if (ConsumerStates.ShardConsumerState.PROCESSING.equals(currentState.state())) { Throwable t = healthCheck(); if (t instanceof Error) { throw (Error) t; } } } @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; }