JobTimeoutTrigger implements StreamProcessorLifecycleAware { void deactivateTimedOutJobs() { final long now = currentTimeMillis(); state.forEachTimedOutEntry( now, (key, record) -> { writer.appendFollowUpCommand( key, JobIntent.TIME_OUT, record, (m) -> m.valueType(ValueType.JOB)); final boolean flushed = writer.flush() >= 0; if (!flushed) { writer.reset(); } return flushed; }); } JobTimeoutTrigger(final JobState state); @Override void onRecovered(final ReadonlyProcessingContext processingContext); @Override void onClose(); @Override void onFailed(); @Override void onPaused(); @Override void onResumed(); static final Duration TIME_OUT_POLLING_INTERVAL; }
@Test public void shouldNotWriteAgainAfterFlushFailed() { when(typedStreamWriter.flush()).thenReturn(1L, -1L); jobTimeoutTrigger.deactivateTimedOutJobs(); final InOrder inOrder = Mockito.inOrder(typedStreamWriter); inOrder .verify(typedStreamWriter) .appendFollowUpCommand(eq(0L), eq(JobIntent.TIME_OUT), any(JobRecord.class), any()); inOrder.verify(typedStreamWriter).flush(); inOrder .verify(typedStreamWriter) .appendFollowUpCommand(eq(1L), eq(JobIntent.TIME_OUT), any(JobRecord.class), any()); inOrder.verify(typedStreamWriter).flush(); inOrder.verify(typedStreamWriter).reset(); inOrder.verifyNoMoreInteractions(); }