@Test public void shouldNotExtendIfMaxReplicaCountIsReached () { when( regionExecPlan.getRegionType() ).thenReturn( PARTITIONED_STATEFUL ); when( regionExecPlan.getReplicaCount() ).thenReturn( MAX_REPLICA_COUNT ); final AdaptationAction action = regionExtender.resolve( regionExecPlan, bottleneckPipelineMetrics ); assertNull( action ); }
RegionRebalanceAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics bottleneckPipelineMetrics ) { checkArgument( execPlan != null ); checkArgument( bottleneckPipelineMetrics != null ); if ( execPlan.getRegionType() != PARTITIONED_STATEFUL ) { return null; } if ( execPlan.getReplicaCount() >= maxReplicaCount ) { LOGGER.warn( "Cannot create a new replica since region: {} has {} replicas", execPlan.getRegionId(), execPlan.getReplicaCount() ); return null; } final int replicaCount = execPlan.getReplicaCount(); return new RegionRebalanceAction( execPlan, replicaCount + 1 ); }
RegionExtender implements BottleneckResolver { RegionRebalanceAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics bottleneckPipelineMetrics ) { checkArgument( execPlan != null ); checkArgument( bottleneckPipelineMetrics != null ); if ( execPlan.getRegionType() != PARTITIONED_STATEFUL ) { return null; } if ( execPlan.getReplicaCount() >= maxReplicaCount ) { LOGGER.warn( "Cannot create a new replica since region: {} has {} replicas", execPlan.getRegionId(), execPlan.getReplicaCount() ); return null; } final int replicaCount = execPlan.getReplicaCount(); return new RegionRebalanceAction( execPlan, replicaCount + 1 ); } }
RegionExtender implements BottleneckResolver { RegionRebalanceAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics bottleneckPipelineMetrics ) { checkArgument( execPlan != null ); checkArgument( bottleneckPipelineMetrics != null ); if ( execPlan.getRegionType() != PARTITIONED_STATEFUL ) { return null; } if ( execPlan.getReplicaCount() >= maxReplicaCount ) { LOGGER.warn( "Cannot create a new replica since region: {} has {} replicas", execPlan.getRegionId(), execPlan.getReplicaCount() ); return null; } final int replicaCount = execPlan.getReplicaCount(); return new RegionRebalanceAction( execPlan, replicaCount + 1 ); } RegionExtender( final int maxReplicaCount ); }
RegionExtender implements BottleneckResolver { RegionRebalanceAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics bottleneckPipelineMetrics ) { checkArgument( execPlan != null ); checkArgument( bottleneckPipelineMetrics != null ); if ( execPlan.getRegionType() != PARTITIONED_STATEFUL ) { return null; } if ( execPlan.getReplicaCount() >= maxReplicaCount ) { LOGGER.warn( "Cannot create a new replica since region: {} has {} replicas", execPlan.getRegionId(), execPlan.getReplicaCount() ); return null; } final int replicaCount = execPlan.getReplicaCount(); return new RegionRebalanceAction( execPlan, replicaCount + 1 ); } RegionExtender( final int maxReplicaCount ); @Override List<Pair<AdaptationAction, List<PipelineId>>> resolve( final RegionExecPlan execPlan, final List<PipelineMetrics> metrics ); }
RegionExtender implements BottleneckResolver { RegionRebalanceAction resolve ( final RegionExecPlan execPlan, final PipelineMetrics bottleneckPipelineMetrics ) { checkArgument( execPlan != null ); checkArgument( bottleneckPipelineMetrics != null ); if ( execPlan.getRegionType() != PARTITIONED_STATEFUL ) { return null; } if ( execPlan.getReplicaCount() >= maxReplicaCount ) { LOGGER.warn( "Cannot create a new replica since region: {} has {} replicas", execPlan.getRegionId(), execPlan.getReplicaCount() ); return null; } final int replicaCount = execPlan.getReplicaCount(); return new RegionRebalanceAction( execPlan, replicaCount + 1 ); } RegionExtender( final int maxReplicaCount ); @Override List<Pair<AdaptationAction, List<PipelineId>>> resolve( final RegionExecPlan execPlan, final List<PipelineMetrics> metrics );  }