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 testConvertToPrimitiveReadsHappen() {
    PipelineOptions deprecatedReadOptions = PipelineOptionsFactory.create();
    deprecatedReadOptions.setRunner(CrashingRunner.class);
    ExperimentalOptions.addExperiment(
        deprecatedReadOptions.as(ExperimentalOptions.class), "use_deprecated_read");

    Pipeline pipeline = Pipeline.create(deprecatedReadOptions);
    pipeline.apply(Read.from(new FakeBoundedSource()));
    pipeline.apply(Read.from(new BoundedToUnboundedSourceAdapter<>(new FakeBoundedSource())));
    SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(pipeline);
    AtomicBoolean sawPrimitiveBoundedRead = new AtomicBoolean();
    AtomicBoolean sawPrimitiveUnboundedRead = new AtomicBoolean();
    pipeline.traverseTopologically(
        new Defaults() {
          @Override
          public CompositeBehavior enterCompositeTransform(Node node) {
            assertThat(node.getTransform(), not(instanceOf(Read.Bounded.class)));
            assertThat(node.getTransform(), not(instanceOf(Read.Unbounded.class)));
            return super.enterCompositeTransform(node);
          }

          @Override
          public void visitPrimitiveTransform(Node node) {
            if (node.getTransform() instanceof SplittableParDo.PrimitiveBoundedRead) {
              sawPrimitiveBoundedRead.set(true);
            } else if (node.getTransform() instanceof SplittableParDo.PrimitiveUnboundedRead) {
              sawPrimitiveUnboundedRead.set(true);
            }
          }
        });
    assertTrue(sawPrimitiveBoundedRead.get());
    assertTrue(sawPrimitiveUnboundedRead.get());
  }