public static SourceOperationResponse performSplit(
      SourceSplitRequest request, PipelineOptions options) throws Exception {
    return performSplitWithApiLimit(
        request, options, DEFAULT_NUM_BUNDLES_LIMIT, DATAFLOW_SPLIT_RESPONSE_API_SIZE_LIMIT);
  }
@Test
  public void testOversplittingDesiredBundleSizeScaledFirst() throws Exception {
    // Create a source that greatly oversplits but with coalescing/compression it would still fit
    // under the API limit. Test that the API limit gets applied first, so oversplitting is
    // reduced.
    com.google.api.services.dataflow.model.Source source =
        translateIOToCloudSource(CountingSource.upTo(8000), options);

    // Without either limit, produces 1000 bundles, total size ~500kb.
    // With only numBundles limit 100, produces 100 bundles, total size ~72kb.
    // With only apiSize limit = 10kb, 72 bundles, total size ~40kb (over the limit but oh well).
    // With numBundles limit 100 and apiSize limit 10kb, should produce 72 bundles.
    // On the other hand, if the numBundles limit of 100 was applied first, we'd get 100 bundles.

    SourceSplitResponse bundledWithOnlyNumBundlesLimit =
        performSplit(
            source, options, 8L, 100 /* numBundles limit */, 10000 * 1024L /* API size limit */);
    assertEquals(100, bundledWithOnlyNumBundlesLimit.getBundles().size());
    assertThat(
        DataflowApiUtils.computeSerializedSizeBytes(bundledWithOnlyNumBundlesLimit),
        greaterThan(10 * 1024L));

    SourceSplitResponse bundledWithOnlySizeLimit =
        performSplit(
            source, options, 8L, 1000000 /* numBundles limit */, 10 * 1024L /* API size limit */);
    int numBundlesWithOnlySizeLimit = bundledWithOnlySizeLimit.getBundles().size();
    assertThat(numBundlesWithOnlySizeLimit, lessThan(100));

    SourceSplitResponse bundledWithSizeLimit = performSplit(source, options, 8L, 100, 10 * 1024L);
    assertEquals(numBundlesWithOnlySizeLimit, bundledWithSizeLimit.getBundles().size());
  }