@Test public void shouldNotResolveBottleneckForSourceRegion () { when( regionDef.isSource() ).thenReturn( true ); final List<AdaptationAction> actions = context.resolveIfBottleneck( bottleneckPredicate, singletonList( bottleneckResolver0 ) ); assertTrue( actions.isEmpty() ); }
List<AdaptationAction> resolveIfBottleneck ( final Predicate<PipelineMetrics> bottleneckPredicate, final List<BottleneckResolver> bottleneckResolvers ) { checkArgument( bottleneckPredicate != null ); checkArgument( bottleneckResolvers != null ); checkState( adaptationActions.isEmpty(), "Region %s cannot try to resolve bottleneck before evaluation of adaptation actions: %s", getRegionId(), adaptationActions ); final List<PipelineId> bottleneckPipelineIds = getBottleneckPipelineIds( bottleneckPredicate ); if ( bottleneckPipelineIds.isEmpty() ) { return emptyList(); } final List<Pair<AdaptationAction, List<PipelineId>>> resolutions = resolveBottlenecks( bottleneckPipelineIds, bottleneckResolvers ); if ( resolutions.isEmpty() ) { nonResolvablePipelineIds.addAll( bottleneckPipelineIds ); adaptationActions = emptyList(); return emptyList(); } adaptationActions = resolutions; currentExecPlan = resolutions.get( resolutions.size() - 1 )._1.getNewExecPlan(); return resolutions.stream().map( Pair::firstElement ).collect( toList() ); }
RegionAdaptationContext { List<AdaptationAction> resolveIfBottleneck ( final Predicate<PipelineMetrics> bottleneckPredicate, final List<BottleneckResolver> bottleneckResolvers ) { checkArgument( bottleneckPredicate != null ); checkArgument( bottleneckResolvers != null ); checkState( adaptationActions.isEmpty(), "Region %s cannot try to resolve bottleneck before evaluation of adaptation actions: %s", getRegionId(), adaptationActions ); final List<PipelineId> bottleneckPipelineIds = getBottleneckPipelineIds( bottleneckPredicate ); if ( bottleneckPipelineIds.isEmpty() ) { return emptyList(); } final List<Pair<AdaptationAction, List<PipelineId>>> resolutions = resolveBottlenecks( bottleneckPipelineIds, bottleneckResolvers ); if ( resolutions.isEmpty() ) { nonResolvablePipelineIds.addAll( bottleneckPipelineIds ); adaptationActions = emptyList(); return emptyList(); } adaptationActions = resolutions; currentExecPlan = resolutions.get( resolutions.size() - 1 )._1.getNewExecPlan(); return resolutions.stream().map( Pair::firstElement ).collect( toList() ); } }
RegionAdaptationContext { List<AdaptationAction> resolveIfBottleneck ( final Predicate<PipelineMetrics> bottleneckPredicate, final List<BottleneckResolver> bottleneckResolvers ) { checkArgument( bottleneckPredicate != null ); checkArgument( bottleneckResolvers != null ); checkState( adaptationActions.isEmpty(), "Region %s cannot try to resolve bottleneck before evaluation of adaptation actions: %s", getRegionId(), adaptationActions ); final List<PipelineId> bottleneckPipelineIds = getBottleneckPipelineIds( bottleneckPredicate ); if ( bottleneckPipelineIds.isEmpty() ) { return emptyList(); } final List<Pair<AdaptationAction, List<PipelineId>>> resolutions = resolveBottlenecks( bottleneckPipelineIds, bottleneckResolvers ); if ( resolutions.isEmpty() ) { nonResolvablePipelineIds.addAll( bottleneckPipelineIds ); adaptationActions = emptyList(); return emptyList(); } adaptationActions = resolutions; currentExecPlan = resolutions.get( resolutions.size() - 1 )._1.getNewExecPlan(); return resolutions.stream().map( Pair::firstElement ).collect( toList() ); } RegionAdaptationContext( final RegionExecPlan currentExecPlan ); }
RegionAdaptationContext { List<AdaptationAction> resolveIfBottleneck ( final Predicate<PipelineMetrics> bottleneckPredicate, final List<BottleneckResolver> bottleneckResolvers ) { checkArgument( bottleneckPredicate != null ); checkArgument( bottleneckResolvers != null ); checkState( adaptationActions.isEmpty(), "Region %s cannot try to resolve bottleneck before evaluation of adaptation actions: %s", getRegionId(), adaptationActions ); final List<PipelineId> bottleneckPipelineIds = getBottleneckPipelineIds( bottleneckPredicate ); if ( bottleneckPipelineIds.isEmpty() ) { return emptyList(); } final List<Pair<AdaptationAction, List<PipelineId>>> resolutions = resolveBottlenecks( bottleneckPipelineIds, bottleneckResolvers ); if ( resolutions.isEmpty() ) { nonResolvablePipelineIds.addAll( bottleneckPipelineIds ); adaptationActions = emptyList(); return emptyList(); } adaptationActions = resolutions; currentExecPlan = resolutions.get( resolutions.size() - 1 )._1.getNewExecPlan(); return resolutions.stream().map( Pair::firstElement ).collect( toList() ); } RegionAdaptationContext( final RegionExecPlan currentExecPlan ); int getRegionId(); RegionDef getRegionDef(); }
RegionAdaptationContext { List<AdaptationAction> resolveIfBottleneck ( final Predicate<PipelineMetrics> bottleneckPredicate, final List<BottleneckResolver> bottleneckResolvers ) { checkArgument( bottleneckPredicate != null ); checkArgument( bottleneckResolvers != null ); checkState( adaptationActions.isEmpty(), "Region %s cannot try to resolve bottleneck before evaluation of adaptation actions: %s", getRegionId(), adaptationActions ); final List<PipelineId> bottleneckPipelineIds = getBottleneckPipelineIds( bottleneckPredicate ); if ( bottleneckPipelineIds.isEmpty() ) { return emptyList(); } final List<Pair<AdaptationAction, List<PipelineId>>> resolutions = resolveBottlenecks( bottleneckPipelineIds, bottleneckResolvers ); if ( resolutions.isEmpty() ) { nonResolvablePipelineIds.addAll( bottleneckPipelineIds ); adaptationActions = emptyList(); return emptyList(); } adaptationActions = resolutions; currentExecPlan = resolutions.get( resolutions.size() - 1 )._1.getNewExecPlan(); return resolutions.stream().map( Pair::firstElement ).collect( toList() ); } RegionAdaptationContext( final RegionExecPlan currentExecPlan ); int getRegionId(); RegionDef getRegionDef();  }