PipelineReplicaRunner implements Runnable { public PipelineReplicaRunnerStatus getStatus () { synchronized ( monitor ) { return status; } } PipelineReplicaRunner( final JokerConfig config,
                            final PipelineReplica pipeline,
                            final Supervisor supervisor,
                            final DownstreamCollector downstreamCollector ); PipelineReplicaRunnerStatus getStatus(); void run();  }
@Test public void shouldCompleteWhileRunning () throws InterruptedException { final CountDownLatch invocationStartLatch = new CountDownLatch( 1 ); final CountDownLatch invocationDoneLatch = new CountDownLatch( 1 ); final TuplesImpl output = new TuplesImpl( 1 ); output.add( new Tuple() ); when( operator.invoke( anyBoolean(), anyObject(), anyObject() ) ).thenAnswer( invocation -> { invocationStartLatch.countDown(); invocationDoneLatch.await( 2, TimeUnit.MINUTES ); return output; } ); thread.start(); reset( supervisor ); pipeline.getCompletionTracker().onStatusChange( operatorDef.getId(), OperatorReplicaStatus.COMPLETED ); invocationStartLatch.await( 2, TimeUnit.MINUTES ); invocationDoneLatch.countDown(); assertTrueEventually( () -> assertEquals( runner.getStatus(), COMPLETED ) ); verify( downstreamCollector ).accept( output ); }
@Test public void shouldCompleteRunningAfterPipelineCompletesItself () { final TuplesImpl output1 = new TuplesImpl( inputOutputPortCount ); output1.add( Tuple.of( "k1", "v1" ) ); final TuplesImpl output2 = new TuplesImpl( inputOutputPortCount ); output2.add( Tuple.of( "k2", "v2" ) ); when( operator.invoke( anyBoolean(), anyObject(), anyObject() ) ).thenReturn( output1, output2 ); thread.start(); sleepUninterruptibly( 1, SECONDS ); pipeline.getCompletionTracker().onStatusChange( operatorDef.getId(), OperatorReplicaStatus.COMPLETED ); assertTrueEventually( () -> assertEquals( runner.getStatus(), COMPLETED ), 10 ); verify( downstreamCollector, atLeastOnce() ).accept( output1 ); verify( downstreamCollector, atLeastOnce() ).accept( output2 ); }
@Test public void shouldCompleteRunningWhenPipelineFailsDuringInvocations () { final RuntimeException failure = new RuntimeException( "expected" ); when( operator.invoke( anyBoolean(), anyObject(), anyObject() ) ).thenThrow( failure ); thread.start(); assertTrueEventually( () -> assertEquals( runner.getStatus(), COMPLETED ), 10 ); verify( supervisor ).notifyPipelineReplicaFailed( id, failure ); }