static SourceOperationResponse performSplitWithApiLimit(
      SourceSplitRequest request, PipelineOptions options, int numBundlesLimit, long apiByteLimit)
      throws Exception {
    // Compute the desired bundle size given by the service, or default if none was provided.
    long desiredBundleSizeBytes = DEFAULT_DESIRED_BUNDLE_SIZE_BYTES;
    SourceSplitOptions splitOptions = request.getOptions();
    if (splitOptions != null && splitOptions.getDesiredBundleSizeBytes() != null) {
      desiredBundleSizeBytes = splitOptions.getDesiredBundleSizeBytes();
    }

    Source<?> anySource = deserializeFromCloudSource(request.getSource().getSpec());
    checkArgument(
        anySource instanceof BoundedSource, "Cannot split a non-Bounded source: %s", anySource);
    return performSplitTyped(
        options,
        (BoundedSource<?>) anySource,
        desiredBundleSizeBytes,
        numBundlesLimit,
        apiByteLimit);
  }
@Test
  public void testSplittingProducedResponseUnderLimit() throws Exception {
    SourceProducingSubSourcesInSplit source = new SourceProducingSubSourcesInSplit(200, 10_000);
    com.google.api.services.dataflow.model.Source cloudSource =
        translateIOToCloudSource(source, options);
    SourceSplitRequest splitRequest = new SourceSplitRequest();
    splitRequest.setSource(cloudSource);

    ExpectedLogs.LogSaver logSaver = new ExpectedLogs.LogSaver();
    LogManager.getLogManager()
        .getLogger("org.apache.beam.runners.dataflow.worker.WorkerCustomSources")
        .addHandler(logSaver);

    WorkerCustomSources.performSplitWithApiLimit(splitRequest, options, 100, 10_000);
    // verify initial split is not valid
    verifyLogged(
        ExpectedLogs.matcher(Level.WARNING, "this is too large for the Google Cloud Dataflow API"),
        logSaver);
    // verify that re-bundle is effective
    verifyLogged(ExpectedLogs.matcher(Level.WARNING, "Re-bundle source"), logSaver);
  }