@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 ); }
public PipelineReplicaRunnerStatus getStatus () { synchronized ( monitor ) { return status; } }
PipelineReplicaRunner implements Runnable { public PipelineReplicaRunnerStatus getStatus () { synchronized ( monitor ) { return status; } } }
PipelineReplicaRunner implements Runnable { public PipelineReplicaRunnerStatus getStatus () { synchronized ( monitor ) { return status; } } PipelineReplicaRunner( final JokerConfig config,
                            final PipelineReplica pipeline,
                            final Supervisor supervisor,
                            final DownstreamCollector downstreamCollector ); }
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(); }
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();  }