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 testSplitAndReadBundlesBack() throws Exception {
    com.google.api.services.dataflow.model.Source source =
        translateIOToCloudSource(CountingSource.upTo(10L), options);
    List<WindowedValue<Integer>> elems = readElemsFromSource(options, source);
    assertEquals(10L, elems.size());
    for (long i = 0; i < 10L; i++) {
      assertEquals(valueInGlobalWindow(i), elems.get((int) i));
    }
    SourceSplitResponse response =
        performSplit(
            source,
            options,
            16L /*desiredBundleSizeBytes for two longs*/,
            null /* numBundles limit */,
            null /* API limit */);
    assertEquals("SOURCE_SPLIT_OUTCOME_SPLITTING_HAPPENED", response.getOutcome());
    List<DerivedSource> bundles = response.getBundles();
    assertEquals(5, bundles.size());
    for (int i = 0; i < 5; ++i) {
      DerivedSource bundle = bundles.get(i);
      assertEquals("SOURCE_DERIVATION_MODE_INDEPENDENT", bundle.getDerivationMode());
      com.google.api.services.dataflow.model.Source bundleSource = bundle.getSource();
      assertTrue(bundleSource.getDoesNotNeedSplitting());
      bundleSource.setCodec(source.getCodec());
      List<WindowedValue<Integer>> xs = readElemsFromSource(options, bundleSource);
      assertThat(
          "Failed on bundle " + i,
          xs,
          contains(valueInGlobalWindow(0L + 2 * i), valueInGlobalWindow(1L + 2 * i)));
      assertTrue(bundle.getSource().getMetadata().getEstimatedSizeBytes() > 0);
    }
  }