@Test public void shouldSplitFusedOperatorOfStatefulRegion () { 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, singletonList( 0 ), 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.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[ 1 ].getOperatorReplicaCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 2 ].getOperatorReplicaCount(), equalTo( 1 ) ); final OperatorReplica pipeline0Operator = newPipelineReplicas[ 0 ].getOperatorReplica( 0 ); final OperatorReplica pipeline1Operator = newPipelineReplicas[ 1 ].getOperatorReplica( 0 ); final OperatorReplica pipeline2Operator = newPipelineReplicas[ 2 ].getOperatorReplica( 0 ); assertThat( pipeline0Operator.getOperatorCount(), equalTo( 1 ) ); assertThat( pipeline1Operator.getOperatorCount(), equalTo( 3 ) ); assertThat( pipeline2Operator.getOperatorCount(), equalTo( 1 ) ); assertThat( newPipelineReplicas[ 0 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 0 ) ) ); assertThat( ( (DefaultOperatorQueue) pipeline0Operator.getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( pipeline0Operator.getQueue() == operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef1.getId(), 0 ) ); assertTrue( pipeline0Operator.getDrainerPool() instanceof BlockingTupleQueueDrainerPool ); assertTrue( newPipelineReplicas[ 0 ].getQueue() instanceof EmptyOperatorQueue ); assertThat( pipeline0Operator.getOperatorDef( 0 ), equalTo( operatorDef1 ) ); assertTrue( pipeline0Operator.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertThat( newPipelineReplicas[ 1 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 1 ) ) ); assertTrue( pipeline1Operator.getQueue() == operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef2.getId(), 0 ) ); assertNull( operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef3.getId(), 0 ) ); assertNull( operatorQueueManager.getDefaultQueue( region.getRegionId(), operatorDef4.getId(), 0 ) ); assertThat( ( (DefaultOperatorQueue) pipeline1Operator.getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( pipeline1Operator.getDrainerPool() instanceof BlockingTupleQueueDrainerPool ); assertTrue( newPipelineReplicas[ 1 ].getQueue() instanceof EmptyOperatorQueue ); assertThat( pipeline1Operator.getOperatorDef( 0 ), equalTo( operatorDef2 ) ); assertThat( pipeline1Operator.getOperatorDef( 1 ), equalTo( operatorDef3 ) ); assertThat( pipeline1Operator.getOperatorDef( 2 ), equalTo( operatorDef4 ) ); assertTrue( pipeline1Operator.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertTrue( pipeline1Operator.getInvocationCtx( 1 ) instanceof FusedInvocationCtx ); assertTrue( pipeline1Operator.getInvocationCtx( 2 ) instanceof FusedInvocationCtx ); assertThat( newPipelineReplicas[ 2 ].id().pipelineId, equalTo( new PipelineId( REGION_ID, 4 ) ) ); assertThat( ( (DefaultOperatorQueue) pipeline2Operator.getQueue() ).getThreadingPref(), equalTo( MULTI_THREADED ) ); assertTrue( pipeline2Operator.getDrainerPool() instanceof BlockingTupleQueueDrainerPool ); assertThat( pipeline2Operator.getOperatorDef( 0 ), equalTo( operatorDef5 ) ); assertTrue( pipeline2Operator.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); 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" ) ); assertNull( pipeline0Operator.getDownstreamCtx() ); assertNull( pipeline1Operator.getDownstreamCtx() ); assertNull( pipeline2Operator.getDownstreamCtx() ); assertThat( newPipelineReplicas[ 0 ].getMeter().getHeadOperatorId(), equalTo( operatorDef1.getId() ) ); assertThat( newPipelineReplicas[ 1 ].getMeter().getHeadOperatorId(), equalTo( operatorDef2.getId() ) ); assertThat( newPipelineReplicas[ 2 ].getMeter().getHeadOperatorId(), equalTo( operatorDef5.getId() ) ); for ( PipelineReplica newPipelineReplica : newPipelineReplicas ) { assertPipelineReplicaMeter( newPipelineReplica ); } }
@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 );  }