PipelineSplitter implements BottleneckResolver { SplitPipelineAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics metrics ) { checkArgument( execPlan != null ); checkArgument( metrics != null ); final PipelineId pipelineId = metrics.getPipelineId(); final int operatorCount = execPlan.getOperatorCountByPipelineStartIndex( pipelineId.getPipelineStartIndex() ); if ( operatorCount < 2 ) { return null; } final int splitIndex = pipelineSplitIndexExtractor.apply( execPlan, metrics ); return splitIndex > 0 ? new SplitPipelineAction( execPlan, pipelineId, splitIndex ) : null; } PipelineSplitter( final BiFunction<RegionExecPlan, PipelineMetrics, Integer> pipelineSplitIndexExtractor ); @Override List<Pair<AdaptationAction, List<PipelineId>>> resolve( final RegionExecPlan execPlan, final List<PipelineMetrics> metrics );  }
@Test public void shouldNotSplitSingleOperatorPipeline () { when( regionExecPlan.getOperatorCountByPipelineStartIndex( pipelineId.getPipelineStartIndex() ) ).thenReturn( 1 ); when( regionExecPlan.withSplitPipeline( singletonList( 0 ) ) ).thenReturn( newRegionExecPlan ); final AdaptationAction action = pipelineSplitter.resolve( regionExecPlan, bottleneckPipelineMetrics ); assertNull( action ); }
@Test public void shouldNotSplitWhenExtractorReturnsNoSplit () { when( regionExecPlan.getOperatorCountByPipelineStartIndex( pipelineId.getPipelineStartIndex() ) ).thenReturn( 2 ); when( pipelineSplitIndexExtractor.apply( regionExecPlan, bottleneckPipelineMetrics ) ).thenReturn( 0 ); final AdaptationAction action = pipelineSplitter.resolve( regionExecPlan, bottleneckPipelineMetrics ); assertNull( action ); }
@Test public void shouldSplitWhenExtractorReturnsSplitIndex () { final int pipelineSplitIndex = 1; when( regionExecPlan.getOperatorCountByPipelineStartIndex( pipelineId.getPipelineStartIndex() ) ).thenReturn( 2 ); when( pipelineSplitIndexExtractor.apply( regionExecPlan, bottleneckPipelineMetrics ) ).thenReturn( pipelineSplitIndex ); when( regionExecPlan.withSplitPipeline( asList( 0, pipelineSplitIndex ) ) ).thenReturn( newRegionExecPlan ); final AdaptationAction action = pipelineSplitter.resolve( regionExecPlan, bottleneckPipelineMetrics ); assertTrue( action instanceof SplitPipelineAction ); final SplitPipelineAction splitPipelineAction = (SplitPipelineAction) action; assertThat( splitPipelineAction.getCurrentExecPlan(), equalTo( regionExecPlan ) ); assertThat( splitPipelineAction.getNewExecPlan(), equalTo( newRegionExecPlan ) ); }