@SuppressWarnings("unchecked") @Test @Ignore public final void testInitializationStateUponFailure() throws Exception { ShardConsumer consumer = new ShardConsumer(recordsPublisher, executorService, shardInfo, logWarningForTaskAfterMillis, shardConsumerArgument, initialState, Function.identity(), 1, taskExecutionListener, 0); when(initialState.createTask(eq(shardConsumerArgument), eq(consumer), any())).thenReturn(initializeTask); when(initializeTask.call()).thenReturn(new TaskResult(new Exception("Bad"))); when(initializeTask.taskType()).thenReturn(TaskType.INITIALIZE); when(initialState.failureTransition()).thenReturn(initialState); CyclicBarrier taskBarrier = new CyclicBarrier(2); when(initialState.requiresDataAvailability()).thenAnswer(i -> { taskBarrier.await(); return false; }); consumer.executeLifecycle(); for (int i = 0; i < 4; ++i) { awaitAndResetBarrier(taskBarrier); } verify(initialState, times(5)).createTask(eq(shardConsumerArgument), eq(consumer), any()); verify(initialState, never()).successTransition(); verify(initialState, never()).shutdownTransition(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; }