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 testSplittingProducedInvalidSource() throws Exception {
    com.google.api.services.dataflow.model.Source cloudSource =
        translateIOToCloudSource(new SourceProducingInvalidSplits("original", null), options);

    expectedException.expect(IllegalArgumentException.class);
    expectedException.expectMessage(
        allOf(
            containsString("Splitting a valid source produced an invalid source"),
            containsString("original"),
            containsString("badBundle")));
    expectedException.expectCause(hasMessage(containsString("intentionally invalid")));
    performSplit(
        cloudSource,
        options,
        null /*desiredBundleSizeBytes*/,
        null /* numBundles limit */,
        null /* API limit */);
  }