@ Override public CompletableFuture < VersionedMetadata < EpochTransitionRecord > > submitScale ( final List < Long > segmentsToSeal , final List < Map . Entry < Double , Double > > newRanges , final long scaleTimestamp , final VersionedMetadata < EpochTransitionRecord > existing ) { return verifyNotSealed ( ) . thenCompose ( v -> { if ( existing == null ) { return getEpochTransition ( ) ; } else { return CompletableFuture . completedFuture ( existing ) ; } } ) . thenCompose ( record -> getActiveEpochRecord ( true ) . thenCompose ( currentEpoch -> { if ( ! record . getObject ( ) . equals ( EpochTransitionRecord . EMPTY ) ) { // verify that it's the same as the supplied input (--> segments to be sealed // and new ranges are identical). else throw scale conflict exception if ( ! RecordHelper . verifyRecordMatchesInput ( segmentsToSeal , newRanges , false , record . getObject ( ) ) ) { log . debug ( "scale conflict, another scale operation is ongoing" ) ; throw new EpochTransitionOperationExceptions . ConflictException ( ) ; } return CompletableFuture . completedFuture ( record ) ; } else { // check input is valid and satisfies preconditions if ( ! RecordHelper . canScaleFor ( segmentsToSeal , currentEpoch ) ) { return updateEpochTransitionNode ( new VersionedMetadata <> ( EpochTransitionRecord . EMPTY , record . getVersion ( ) ) ) . thenApply ( x -> { log . warn ( "scale precondition failed {}" , segmentsToSeal ) ; throw new EpochTransitionOperationExceptions . PreConditionFailureException ( ) ; } ) ; } if ( ! RecordHelper . validateInputRange ( segmentsToSeal , newRanges , currentEpoch ) ) { log . error ( "scale input invalid {} {}" , segmentsToSeal , newRanges ) ; throw new EpochTransitionOperationExceptions . InputInvalidException ( ) ; } EpochTransitionRecord epochTransition = RecordHelper . computeEpochTransition ( currentEpoch , segmentsToSeal , newRanges , scaleTimestamp ) ; return updateEpochTransitionNode ( new VersionedMetadata <> ( epochTransition , record . getVersion ( ) ) ) . thenApply ( version -> { log . info ( "scale for stream {}/{} accepted. Segments to seal = {}" , scope , name , epochTransition . getSegmentsToSeal ( ) ) ; return new VersionedMetadata <> ( epochTransition , version ) ; } ) ; } } ) ) ; }
This method attempts to start a new scale workflow . For this it first computes epoch transition and stores it in the metadastore . This method can be called by manual scale or during the processing of auto - scale event . Which means there could be concurrent calls to this method .