@Test public void shouldMergeAllPipelinesOfStatefulRegionWithFusibleOperator () { final OperatorDef operatorDef0 = OperatorDefBuilder.newInstance( "op0", StatelessInput0Output1Operator.class ).build(); final OperatorDef operatorDef1 = OperatorDefBuilder.newInstance( "op1", FusibleStatefulInput1Output1Operator.class ).build(); final OperatorDef operatorDef2 = OperatorDefBuilder.newInstance( "op2", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef3 = OperatorDefBuilder.newInstance( "op3", FusibleStatefulInput1Output1Operator.class ).build(); final OperatorDef operatorDef4 = OperatorDefBuilder.newInstance( "op4", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef5 = OperatorDefBuilder.newInstance( "op5", FusibleStatefulInput1Output1Operator.class ).build(); final FlowDef flow = new FlowDefBuilder().add( operatorDef0 ) .add( operatorDef1 ) .add( operatorDef2 ) .add( operatorDef3 ) .add( operatorDef4 ) .add( operatorDef5 ) .connect( "op0", "op1" ) .connect( "op1", "op2" ) .connect( "op2", "op3" ) .connect( "op3", "op4" ) .connect( "op4", "op5" ) .build(); final RegionDef regionDef = new RegionDef( REGION_ID, STATEFUL, emptyList(), asList( operatorDef1, operatorDef2, operatorDef3, operatorDef4, operatorDef5 ) ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, asList( 0, 2, 4 ), 1 ); final Region region = regionManager.createRegion( flow, regionExecPlan ); initialize( region ); operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef1.getId() ).getKVStore( null ).set( "key1", "val1" ); operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef3.getId() ).getKVStore( null ).set( "key3", "val3" ); operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef5.getId() ).getKVStore( null ).set( "key5", "val5" ); final Region newRegion = pipelineTransformer.mergePipelines( region, asList( 0, 2, 4 ) ); assertThat( newRegion.getExecPlan().getReplicaCount(), equalTo( 1 ) ); assertThat( newRegion.getExecPlan().getPipelineStartIndices(), equalTo( singletonList( 0 ) ) ); final PipelineReplica[] newPipelineReplicas = newRegion.getReplicaPipelines( 0 ); assertThat( newPipelineReplicas.length, equalTo( 1 ) ); final PipelineReplica newPipelineReplica = newPipelineReplicas[ 0 ]; assertThat( newPipelineReplica.getOperatorReplicaCount(), equalTo( 1 ) ); final OperatorReplica pipelineOperator0 = newPipelineReplica.getOperatorReplica( 0 ); assertThat( pipelineOperator0.getOperatorCount(), equalTo( 5 ) ); assertThat( pipelineOperator0.getOperatorDef( 0 ), equalTo( operatorDef1 ) ); assertThat( pipelineOperator0.getOperatorDef( 1 ), equalTo( operatorDef2 ) ); assertThat( pipelineOperator0.getOperatorDef( 2 ), equalTo( operatorDef3 ) ); assertThat( pipelineOperator0.getOperatorDef( 3 ), equalTo( operatorDef4 ) ); assertThat( pipelineOperator0.getOperatorDef( 4 ), equalTo( operatorDef5 ) ); assertTrue( pipelineOperator0.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertTrue( pipelineOperator0.getInvocationCtx( 1 ) instanceof FusedInvocationCtx ); assertTrue( pipelineOperator0.getInvocationCtx( 2 ) instanceof FusedInvocationCtx ); assertTrue( pipelineOperator0.getInvocationCtx( 3 ) instanceof FusedInvocationCtx ); assertTrue( pipelineOperator0.getInvocationCtx( 4 ) instanceof FusedInvocationCtx ); assertThat( ( (DefaultOperatorQueue) pipelineOperator0.getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( pipelineOperator0.getQueue() == operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef1.getId(), 0 ) ); assertTrue( pipelineOperator0.getDrainerPool() instanceof BlockingTupleQueueDrainerPool ); assertTrue( newPipelineReplica.getQueue() instanceof EmptyOperatorQueue ); assertThat( operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef1.getId() ).getKVStore( null ).get( "key1" ), equalTo( "val1" ) ); assertThat( operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef3.getId() ).getKVStore( null ).get( "key3" ), equalTo( "val3" ) ); assertThat( operatorKVStoreManager.getDefaultKVStore( region.getRegionId(), operatorDef5.getId() ).getKVStore( null ).get( "key5" ), equalTo( "val5" ) ); final UpstreamCtx[] upstreamCtxes = region.getUpstreamCtxes(); assertTrue( pipelineOperator0.getUpstreamCtx( 0 ) == upstreamCtxes[ 0 ] ); assertTrue( pipelineOperator0.getUpstreamCtx( 1 ) == upstreamCtxes[ 1 ] ); assertTrue( pipelineOperator0.getUpstreamCtx( 2 ) == upstreamCtxes[ 2 ] ); assertTrue( pipelineOperator0.getUpstreamCtx( 3 ) == upstreamCtxes[ 3 ] ); assertTrue( pipelineOperator0.getUpstreamCtx( 4 ) == upstreamCtxes[ 4 ] ); assertNull( pipelineOperator0.getDownstreamCtx() ); final PipelineReplicaMeter meter = newPipelineReplica.getMeter(); assertThat( meter.getHeadOperatorId(), equalTo( pipelineOperator0.getOperatorDef( 0 ).getId() ) ); assertThat( meter.getInputPortCount(), equalTo( pipelineOperator0.getOperatorDef( 0 ).getInputPortCount() ) ); assertPipelineReplicaMeter( newPipelineReplica ); }
@Override public Region mergePipelines ( Region region, final List<Integer> pipelineStartIndicesToMerge ) { final int upPipelineStartIndex = pipelineStartIndicesToMerge.get( 0 ); for ( int downPipelineStartIndex : pipelineStartIndicesToMerge.subList( 1, pipelineStartIndicesToMerge.size() ) ) { region = mergePipelines( region, upPipelineStartIndex, downPipelineStartIndex ); } return region; }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region mergePipelines ( Region region, final List<Integer> pipelineStartIndicesToMerge ) { final int upPipelineStartIndex = pipelineStartIndicesToMerge.get( 0 ); for ( int downPipelineStartIndex : pipelineStartIndicesToMerge.subList( 1, pipelineStartIndicesToMerge.size() ) ) { region = mergePipelines( region, upPipelineStartIndex, downPipelineStartIndex ); } return region; } }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region mergePipelines ( Region region, final List<Integer> pipelineStartIndicesToMerge ) { final int upPipelineStartIndex = pipelineStartIndicesToMerge.get( 0 ); for ( int downPipelineStartIndex : pipelineStartIndicesToMerge.subList( 1, pipelineStartIndicesToMerge.size() ) ) { region = mergePipelines( region, upPipelineStartIndex, downPipelineStartIndex ); } return region; } @Inject  PipelineTransformerImpl( final JokerConfig config,
                                     final PartitionService partitionService,
                                     final OperatorQueueManager operatorQueueManager,
                                     final OperatorKVStoreManager operatorKVStoreManager,
                                     final PartitionKeyExtractorFactory partitionKeyExtractorFactory ); }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region mergePipelines ( Region region, final List<Integer> pipelineStartIndicesToMerge ) { final int upPipelineStartIndex = pipelineStartIndicesToMerge.get( 0 ); for ( int downPipelineStartIndex : pipelineStartIndicesToMerge.subList( 1, pipelineStartIndicesToMerge.size() ) ) { region = mergePipelines( region, upPipelineStartIndex, downPipelineStartIndex ); } return region; } @Inject  PipelineTransformerImpl( final JokerConfig config,
                                     final PartitionService partitionService,
                                     final OperatorQueueManager operatorQueueManager,
                                     final OperatorKVStoreManager operatorKVStoreManager,
                                     final PartitionKeyExtractorFactory partitionKeyExtractorFactory ); @Override Region mergePipelines( Region region, final List<Integer> pipelineStartIndicesToMerge ); Region mergePipelines( final Region region, final int upPipelineStartIndex, final int downPipelineStartIndex ); @Override Region splitPipeline( Region region, final List<Integer> pipelineStartIndicesToSplit ); Region splitPipeline( final Region region, final int pipelineStartIndex, final int splitPipelineStartIndex ); }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region mergePipelines ( Region region, final List<Integer> pipelineStartIndicesToMerge ) { final int upPipelineStartIndex = pipelineStartIndicesToMerge.get( 0 ); for ( int downPipelineStartIndex : pipelineStartIndicesToMerge.subList( 1, pipelineStartIndicesToMerge.size() ) ) { region = mergePipelines( region, upPipelineStartIndex, downPipelineStartIndex ); } return region; } @Inject  PipelineTransformerImpl( final JokerConfig config,
                                     final PartitionService partitionService,
                                     final OperatorQueueManager operatorQueueManager,
                                     final OperatorKVStoreManager operatorKVStoreManager,
                                     final PartitionKeyExtractorFactory partitionKeyExtractorFactory ); @Override Region mergePipelines( Region region, final List<Integer> pipelineStartIndicesToMerge ); Region mergePipelines( final Region region, final int upPipelineStartIndex, final int downPipelineStartIndex ); @Override Region splitPipeline( Region region, final List<Integer> pipelineStartIndicesToSplit ); Region splitPipeline( final Region region, final int pipelineStartIndex, final int splitPipelineStartIndex );  }