@Test public void shouldCopyNonSplitPipelinesAtHeadOfTheRegion () { final OperatorDef operatorDef0 = OperatorDefBuilder.newInstance( "op0", StatefulInput0Output1Operator.class ).build(); final OperatorDef operatorDef1 = OperatorDefBuilder.newInstance( "op1", PartitionedStatefulInput1Output1Operator.class ) .setPartitionFieldNames( singletonList( "field" ) ) .build(); final OperatorDef operatorDef2 = OperatorDefBuilder.newInstance( "op2", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef3 = OperatorDefBuilder.newInstance( "op3", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef4 = OperatorDefBuilder.newInstance( "op4", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef5 = OperatorDefBuilder.newInstance( "op5", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef6 = OperatorDefBuilder.newInstance( "op6", StatelessInput1Output1Operator.class ).build(); final FlowDef flow = new FlowDefBuilder().add( operatorDef0 ) .add( operatorDef1 ) .add( operatorDef2 ) .add( operatorDef3 ) .add( operatorDef4 ) .add( operatorDef5 ) .add( operatorDef6 ) .connect( "op0", "op1" ) .connect( "op1", "op2" ) .connect( "op2", "op3" ) .connect( "op3", "op4" ) .connect( "op4", "op5" ) .connect( "op5", "op6" ) .build(); final RegionDef regionDef = new RegionDef( REGION_ID, PARTITIONED_STATEFUL, singletonList( "field" ), asList( operatorDef1, operatorDef2, operatorDef3, operatorDef4, operatorDef5, operatorDef6 ) ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, asList( 0, 1, 2 ), 2 ); final Region region = regionManager.createRegion( flow, regionExecPlan ); initialize( region ); final Region newRegion = pipelineTransformer.splitPipeline( region, asList( 2, 4 ) ); assertThat( newRegion.getPipelineReplicas( 0 ), equalTo( region.getPipelineReplicas( 0 ) ) ); assertThat( newRegion.getPipelineReplicas( 1 ), equalTo( region.getPipelineReplicas( 1 ) ) ); }
@Override public Region splitPipeline ( Region region, final List<Integer> pipelineStartIndicesToSplit ) { final int pipelineStartIndex = pipelineStartIndicesToSplit.get( 0 ); final List<Integer> splitIndices = new ArrayList<>( pipelineStartIndicesToSplit.subList( 1, pipelineStartIndicesToSplit.size() ) ); reverse( splitIndices ); for ( int splitPipelineStartIndex : splitIndices ) { region = splitPipeline( region, pipelineStartIndex, splitPipelineStartIndex ); } return region; }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region splitPipeline ( Region region, final List<Integer> pipelineStartIndicesToSplit ) { final int pipelineStartIndex = pipelineStartIndicesToSplit.get( 0 ); final List<Integer> splitIndices = new ArrayList<>( pipelineStartIndicesToSplit.subList( 1, pipelineStartIndicesToSplit.size() ) ); reverse( splitIndices ); for ( int splitPipelineStartIndex : splitIndices ) { region = splitPipeline( region, pipelineStartIndex, splitPipelineStartIndex ); } return region; } }
PipelineTransformerImpl implements PipelineTransformer { @Override public Region splitPipeline ( Region region, final List<Integer> pipelineStartIndicesToSplit ) { final int pipelineStartIndex = pipelineStartIndicesToSplit.get( 0 ); final List<Integer> splitIndices = new ArrayList<>( pipelineStartIndicesToSplit.subList( 1, pipelineStartIndicesToSplit.size() ) ); reverse( splitIndices ); for ( int splitPipelineStartIndex : splitIndices ) { region = splitPipeline( region, pipelineStartIndex, splitPipelineStartIndex ); } 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 splitPipeline ( Region region, final List<Integer> pipelineStartIndicesToSplit ) { final int pipelineStartIndex = pipelineStartIndicesToSplit.get( 0 ); final List<Integer> splitIndices = new ArrayList<>( pipelineStartIndicesToSplit.subList( 1, pipelineStartIndicesToSplit.size() ) ); reverse( splitIndices ); for ( int splitPipelineStartIndex : splitIndices ) { region = splitPipeline( region, pipelineStartIndex, splitPipelineStartIndex ); } 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 splitPipeline ( Region region, final List<Integer> pipelineStartIndicesToSplit ) { final int pipelineStartIndex = pipelineStartIndicesToSplit.get( 0 ); final List<Integer> splitIndices = new ArrayList<>( pipelineStartIndicesToSplit.subList( 1, pipelineStartIndicesToSplit.size() ) ); reverse( splitIndices ); for ( int splitPipelineStartIndex : splitIndices ) { region = splitPipeline( region, pipelineStartIndex, splitPipelineStartIndex ); } 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 );  }