@Test public void shouldSplitSinglePipelineOfPartitionedStatefulRegion () { final OperatorDef operatorDef0 = OperatorDefBuilder.newInstance( "op0", StatelessInput0Output1Operator.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", PartitionedStatefulInput1Output1Operator.class ) .setPartitionFieldNames( singletonList( "field" ) ) .build(); final OperatorDef operatorDef4 = OperatorDefBuilder.newInstance( "op4", StatelessInput1Output1Operator.class ).build(); final OperatorDef operatorDef5 = OperatorDefBuilder.newInstance( "op5", PartitionedStatefulInput1Output1Operator.class ) .setPartitionFieldNames( singletonList( "field" ) ) .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, PARTITIONED_STATEFUL, singletonList( "field" ), asList( operatorDef1, operatorDef2, operatorDef3, operatorDef4, operatorDef5 ) ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, singletonList( 0 ), 1 ); final Region region = regionManager.createRegion( flow, regionExecPlan ); initialize( region ); final Region newRegion = pipelineTransformer.splitPipeline( region, asList( 0, 1, 4 ) ); assertThat( newRegion.getExecPlan().getReplicaCount(), equalTo( 1 ) ); assertThat( newRegion.getExecPlan().getPipelineStartIndices(), equalTo( asList( 0, 1, 4 ) ) ); final PipelineReplica[] newPipelineReplicas = newRegion.getReplicaPipelines( 0 ); assertThat( newPipelineReplicas[ 0 ].getOperatorReplicaCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 0 ].getOperatorCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 1 ].getOperatorReplicaCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 1 ].getOperatorCount(), equalTo( 3 ) ); assertThat( newPipelineReplicas[ 2 ].getOperatorReplicaCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 2 ].getOperatorCount(), equalTo( 1 ) ); final OperatorReplica pipelineOperator0 = newPipelineReplicas[ 0 ].getOperatorReplica( 0 ); final OperatorReplica pipelineOperator1 = newPipelineReplicas[ 1 ].getOperatorReplica( 0 ); final OperatorReplica pipelineOperator2 = newPipelineReplicas[ 2 ].getOperatorReplica( 0 ); assertThat( newPipelineReplicas[ 0 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 0 ) ) ); assertTrue( newPipelineReplicas[ 0 ].getQueue() instanceof DefaultOperatorQueue ); assertThat( ( (DefaultOperatorQueue) newPipelineReplicas[ 0 ].getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( newPipelineReplicas[ 0 ].getQueue() == operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef1.getId(), 0 ) ); assertTrue( pipelineOperator0.getQueue() instanceof PartitionedOperatorQueue ); assertTrue( pipelineOperator0.getQueue() == operatorQueueManager.getPartitionedQueueOrFail( region.getRegionId(), operatorDef1.getId(), 0 ) ); assertTrue( pipelineOperator0.getDrainerPool() instanceof NonBlockingTupleQueueDrainerPool ); assertThat( pipelineOperator0.getOperatorDef( 0 ), equalTo( operatorDef1 ) ); assertTrue( pipelineOperator0.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertThat( newPipelineReplicas[ 1 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 1 ) ) ); assertTrue( newPipelineReplicas[ 1 ].getQueue() instanceof EmptyOperatorQueue ); assertThat( ( (DefaultOperatorQueue) pipelineOperator1.getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( pipelineOperator1.getDrainerPool() instanceof BlockingTupleQueueDrainerPool ); assertNull( operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef3.getId(), 0 ) ); assertNull( operatorQueueManager.getPartitionedQueues( region.getRegionId(), operatorDef3.getId() ) ); assertNull( operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef4.getId(), 0 ) ); assertThat( pipelineOperator1.getOperatorDef( 0 ), equalTo( operatorDef2 ) ); assertThat( pipelineOperator1.getOperatorDef( 1 ), equalTo( operatorDef3 ) ); assertThat( pipelineOperator1.getOperatorDef( 2 ), equalTo( operatorDef4 ) ); assertTrue( pipelineOperator1.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertTrue( pipelineOperator1.getInvocationCtx( 1 ) instanceof FusedPartitionedInvocationCtx ); assertTrue( pipelineOperator1.getInvocationCtx( 2 ) instanceof FusedInvocationCtx ); assertThat( newPipelineReplicas[ 2 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 4 ) ) ); assertTrue( newPipelineReplicas[ 2 ].getQueue() instanceof DefaultOperatorQueue ); assertThat( ( (DefaultOperatorQueue) newPipelineReplicas[ 2 ].getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( newPipelineReplicas[ 2 ].getQueue() == operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef5.getId(), 0 ) ); assertTrue( pipelineOperator2.getQueue() instanceof PartitionedOperatorQueue ); assertTrue( pipelineOperator2.getDrainerPool() instanceof NonBlockingTupleQueueDrainerPool ); assertThat( pipelineOperator2.getOperatorDef( 0 ), equalTo( operatorDef5 ) ); assertTrue( pipelineOperator2.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertNull( pipelineOperator0.getDownstreamCtx() ); assertNull( pipelineOperator1.getDownstreamCtx() ); assertNull( pipelineOperator2.getDownstreamCtx() ); final PipelineReplicaMeter meter0 = newPipelineReplicas[ 0 ].getMeter(); assertThat( meter0.getHeadOperatorId(), equalTo( pipelineOperator0.getOperatorDef( 0 ).getId() ) ); assertThat( meter0.getInputPortCount(), equalTo( pipelineOperator0.getOperatorDef( 0 ).getInputPortCount() ) ); final PipelineReplicaMeter meter1 = newPipelineReplicas[ 1 ].getMeter(); assertThat( meter1.getHeadOperatorId(), equalTo( pipelineOperator1.getOperatorDef( 0 ).getId() ) ); assertThat( meter1.getInputPortCount(), equalTo( pipelineOperator1.getOperatorDef( 0 ).getInputPortCount() ) ); final PipelineReplicaMeter meter2 = newPipelineReplicas[ 2 ].getMeter(); assertThat( meter2.getHeadOperatorId(), equalTo( pipelineOperator2.getOperatorDef( 0 ).getId() ) ); assertThat( meter2.getInputPortCount(), equalTo( pipelineOperator2.getOperatorDef( 0 ).getInputPortCount() ) ); for ( PipelineReplica p : newPipelineReplicas ) { assertPipelineReplicaMeter( p ); } }
@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 );  }