public static void convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(Pipeline pipeline) {
    if (!(ExperimentalOptions.hasExperiment(pipeline.getOptions(), "use_sdf_read")
            || ExperimentalOptions.hasExperiment(
                pipeline.getOptions(), "use_unbounded_sdf_wrapper"))
        || ExperimentalOptions.hasExperiment(
            pipeline.getOptions(), "beam_fn_api_use_deprecated_read")
        || ExperimentalOptions.hasExperiment(pipeline.getOptions(), "use_deprecated_read")) {
      convertReadBasedSplittableDoFnsToPrimitiveReads(pipeline);
    }
  }
@Test
  public void testConvertIsSkippedWhenUsingUseSDFRead() {
    PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
    pipelineOptions.setRunner(CrashingRunner.class);
    ExperimentalOptions.addExperiment(
        pipelineOptions.as(ExperimentalOptions.class), "use_sdf_read");
    Pipeline sdfRead = Pipeline.create(pipelineOptions);
    sdfRead.apply(Read.from(new FakeBoundedSource()));
    sdfRead.apply(Read.from(new BoundedToUnboundedSourceAdapter<>(new FakeBoundedSource())));
    SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(sdfRead);
    sdfRead.traverseTopologically(
        new Defaults() {
          @Override
          public void visitPrimitiveTransform(Node node) {
            assertThat(
                node.getTransform(), not(instanceOf(SplittableParDo.PrimitiveBoundedRead.class)));
            assertThat(
                node.getTransform(), not(instanceOf(SplittableParDo.PrimitiveUnboundedRead.class)));
          }
        });
  }