Fix to prevent invalid ShardConsumer state transitions due to rejected executions #560
Conversation
…task execution is rejected by the executor service.
| mockSuccessfulProcessing(null); | ||
|
|
||
| // Failing the initialization task and all other attempts after that. | ||
| doCallRealMethod() |
There was a problem hiding this comment.
Isn't this test supposed to test the behavior of the ShardConsumer on failures when executorservice is not able to handle the submissions of runnables? Testing multithreaded behavior in unit tests is generally not recommended. Would mocking the executorservice not work in this scenario?
There was a problem hiding this comment.
Multithreaded behavior is not tested here. Using directexecutor instead.
| do { | ||
| try { | ||
| consumer.executeLifecycle(); | ||
| Thread.sleep(100); |
There was a problem hiding this comment.
Avoid thread sleeps in unit tests.
| // Suppress any exception like the scheduler. | ||
| } | ||
| } while (--arbitraryExecutionCount > 0); | ||
| assertEquals(ShardConsumerState.INITIALIZING.consumerState().state(), consumer.currentState().state()); |
There was a problem hiding this comment.
You may want to validate the number of calls to the executorservice.
There was a problem hiding this comment.
Added the expectation on number of calls
| mockSuccessfulProcessing(null); | ||
|
|
||
| int arbitraryExecutionCount = 3; | ||
| do { |
There was a problem hiding this comment.
Consider writing as regular while loop for readability
| do { | ||
| try { | ||
| consumer.executeLifecycle(); | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
Should the test not fail if an exception is thrown?
| .when(failingService).execute(any()); | ||
|
|
||
| int arbitraryExecutionCount = 5; | ||
| do { |
There was a problem hiding this comment.
No need to change: Out of curiosity, specific reason for do..while instead of for loop?
| .doThrow(new RejectedExecutionException()) | ||
| .doThrow(new RejectedExecutionException()) | ||
| .doThrow(new RejectedExecutionException()) | ||
| .doAnswer(invocation -> directlyExecuteRunnable(invocation)) |
There was a problem hiding this comment.
Is it not possible to use ArgumentCaptor here?
Issue #, if available:
Description of changes:
The fix is to not propagate the RejectedExecutionException from ShardConsumer to the Scheduler and mark the execution as failure at the ShardConsumer. This will not allow successful state transition to happen.
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.