code: public CompletableFuture<ScaleResponse> manualScale(String scope, String stream, List<Long> segmentsToSeal,
                                                          List<Map.Entry<Double, Double>> newRanges, long scaleTimestamp,
                                                          OperationContext context) {
          final long requestId = requestTracker.getRequestIdFor("scaleStream", scope, stream, String.valueOf(scaleTimestamp));
          ScaleOpEvent event = new ScaleOpEvent(scope, stream, segmentsToSeal, newRanges, true, scaleTimestamp, requestId);
  
          return addIndexAndSubmitTask(event,
                  () -> streamMetadataStore.submitScale(scope, stream, segmentsToSeal, new ArrayList<>(newRanges),
                          scaleTimestamp, null, context, executor))
                          .handle((startScaleResponse, e) -> {
                              ScaleResponse.Builder response = ScaleResponse.newBuilder();
  
                              if (e != null) {
                                  Throwable cause = Exceptions.unwrap(e);
                                  if (cause instanceof EpochTransitionOperationExceptions.PreConditionFailureException) {
                                      response.setStatus(ScaleResponse.ScaleStreamStatus.PRECONDITION_FAILED);
                                  } else {
                                      log.warn(requestId, "Scale for stream {}/{} failed with exception {}", scope, stream, cause);
                                      response.setStatus(ScaleResponse.ScaleStreamStatus.FAILURE);
                                  }
                              } else {
                                  log.info(requestId, "scale for stream {}/{} started successfully", scope, stream);
                                  response.setStatus(ScaleResponse.ScaleStreamStatus.STARTED);
                                  response.addAllSegments(
                                          startScaleResponse.getObject().getNewSegmentsWithRange().entrySet()
                                                  .stream()
                                                  .map(segment -> convert(scope, stream, segment))
                                                  .collect(Collectors.toList()));
                                  response.setEpoch(startScaleResponse.getObject().getActiveEpoch());
                              }
                              return response.build();
                          });
      }
docs: Helper method to perform scale operation against an scale request.
  This method posts a request in the request stream and then starts the scale operation while
  tracking it's progress. Eventually, after scale completion, it sends a response to the caller.
  
  @param scope          scope.
  @param stream         stream name.
  @param segmentsToSeal segments to be sealed.
  @param newRanges      key ranges for new segments.
  @param scaleTimestamp scaling time stamp.
  @param context        optional context
  @return returns the newly created segments.