StreamProcessor extends Actor implements HealthMonitorable { private void onRecovered(final long lastReprocessedPosition) { phase = Phase.PROCESSING; onCommitPositionUpdatedCondition = actor.onCondition( getName() + "-on-commit-position-updated", processingStateMachine::readNextEvent); logStream.registerOnCommitPositionUpdatedCondition(onCommitPositionUpdatedCondition); lifecycleAwareListeners.forEach(l -> l.onRecovered(processingContext)); processingStateMachine.startProcessing(lastReprocessedPosition); } protected  StreamProcessor(final StreamProcessorBuilder processorBuilder); static StreamProcessorBuilder builder(); @Override String getName(); @Override ActorFuture<Void> closeAsync(); @Override void onActorFailed(); ActorFuture<Void> openAsync(); boolean isOpened(); boolean isClosed(); boolean isFailed(); ActorFuture<Long> getLastProcessedPositionAsync(); ActorFuture<Long> getLastWrittenPositionAsync(); @Override HealthStatus getHealthStatus(); @Override void addFailureListener(final FailureListener failureListener); ActorFuture<Phase> getCurrentPhase(); ActorFuture<Void> pauseProcessing(); void resumeProcessing(); static final long UNSET_POSITION; static final Duration HEALTH_CHECK_TICK_DURATION; }
@Test public void shouldCallStreamProcessorLifecycle() throws Exception { final StreamProcessorLifecycleAware lifecycleAware = mock(StreamProcessorLifecycleAware.class); final CountDownLatch recoveredLatch = new CountDownLatch(1); streamProcessorRule.startTypedStreamProcessor( (processors, state) -> processors .withListener(lifecycleAware) .withListener( new StreamProcessorLifecycleAware() { @Override public void onRecovered(final ReadonlyProcessingContext context) { recoveredLatch.countDown(); } })); recoveredLatch.await(); streamProcessorRule.closeStreamProcessor(); final InOrder inOrder = inOrder(lifecycleAware); inOrder.verify(lifecycleAware, times(1)).onRecovered(any()); inOrder.verify(lifecycleAware, times(1)).onClose(); inOrder.verifyNoMoreInteractions(); }
@Test public void shouldCallRecordProcessorLifecycle() throws Exception { final var typedRecordProcessor = mock(TypedRecordProcessor.class); final var recoveredLatch = new CountDownLatch(1); streamProcessorRule.startTypedStreamProcessor( (processors, state) -> processors .onEvent(ValueType.DEPLOYMENT, DeploymentIntent.CREATE, typedRecordProcessor) .withListener( new StreamProcessorLifecycleAware() { @Override public void onRecovered(final ReadonlyProcessingContext context) { recoveredLatch.countDown(); } })); recoveredLatch.await(); streamProcessorRule.closeStreamProcessor(); final InOrder inOrder = inOrder(typedRecordProcessor); inOrder.verify(typedRecordProcessor, times(1)).onRecovered(any()); inOrder.verify(typedRecordProcessor, times(1)).onClose(); inOrder.verifyNoMoreInteractions(); }
@Test public void shouldProcessRecord() { final TypedRecordProcessor<?> typedRecordProcessor = mock(TypedRecordProcessor.class); streamProcessorRule.startTypedStreamProcessor( (processors, state) -> processors.onEvent( ValueType.WORKFLOW_INSTANCE, WorkflowInstanceIntent.ELEMENT_ACTIVATING, typedRecordProcessor)); final long position = streamProcessorRule.writeWorkflowInstanceEvent(WorkflowInstanceIntent.ELEMENT_ACTIVATING); final InOrder inOrder = inOrder(typedRecordProcessor); inOrder.verify(typedRecordProcessor, TIMEOUT.times(1)).onRecovered(any()); inOrder .verify(typedRecordProcessor, TIMEOUT.times(1)) .processRecord(eq(position), any(), any(), any(), any()); inOrder.verifyNoMoreInteractions(); Assertions.assertThat( streamProcessorRule.getZeebeState().getLastSuccessfulProcessedRecordPosition()) .isEqualTo(position); }
@Test public void shouldRetryProcessingRecordOnRecoverableException() { final TypedRecordProcessor<?> typedRecordProcessor = mock(TypedRecordProcessor.class); final AtomicInteger count = new AtomicInteger(0); doAnswer( (invocationOnMock -> { if (count.getAndIncrement() == 0) { throw new RecoverableException("recoverable"); } return null; })) .when(typedRecordProcessor) .processRecord(anyLong(), any(), any(), any(), any()); streamProcessorRule.startTypedStreamProcessor( (processors, state) -> processors.onEvent( ValueType.WORKFLOW_INSTANCE, WorkflowInstanceIntent.ELEMENT_ACTIVATING, typedRecordProcessor)); final long position = streamProcessorRule.writeWorkflowInstanceEvent(WorkflowInstanceIntent.ELEMENT_ACTIVATING); final InOrder inOrder = inOrder(typedRecordProcessor); inOrder.verify(typedRecordProcessor, TIMEOUT.times(1)).onRecovered(any()); inOrder .verify(typedRecordProcessor, TIMEOUT.times(2)) .processRecord(eq(position), any(), any(), any(), any()); inOrder.verifyNoMoreInteractions(); }
@Test public void shouldIgnoreRecordWhenNoProcessorExistForThisType() { final TypedRecordProcessor<?> typedRecordProcessor = mock(TypedRecordProcessor.class); streamProcessorRule.startTypedStreamProcessor( (processors, state) -> processors.onEvent( ValueType.WORKFLOW_INSTANCE, WorkflowInstanceIntent.ELEMENT_ACTIVATING, typedRecordProcessor)); final long firstPosition = streamProcessorRule.writeWorkflowInstanceEvent(WorkflowInstanceIntent.ELEMENT_ACTIVATING); final long secondPosition = streamProcessorRule.writeWorkflowInstanceEvent(WorkflowInstanceIntent.ELEMENT_ACTIVATED); final InOrder inOrder = inOrder(typedRecordProcessor); inOrder.verify(typedRecordProcessor, TIMEOUT.times(1)).onRecovered(any()); inOrder .verify(typedRecordProcessor, TIMEOUT.times(1)) .processRecord(eq(firstPosition), any(), any(), any(), any()); inOrder .verify(typedRecordProcessor, never()) .processRecord(eq(secondPosition), any(), any(), any(), any()); inOrder.verifyNoMoreInteractions(); }