@Override
  public SparkPipelineResult run(Pipeline pipeline) {
    boolean isStreaming =
        options.isStreaming() || options.as(TestSparkPipelineOptions.class).isForceStreaming();

    // Default to using the primitive versions of Read.Bounded and Read.Unbounded.
    // TODO(https://github.com/apache/beam/issues/20530): Use SDF read as default when we address
    // performance issue.
    if (!ExperimentalOptions.hasExperiment(pipeline.getOptions(), "beam_fn_api")) {
      SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(pipeline);
    }

    JavaSparkContext jsc =
        SparkContextFactory.getSparkContext(pipeline.getOptions().as(SparkPipelineOptions.class));
    JavaStreamingContext jssc =
        new JavaStreamingContext(jsc, new org.apache.spark.streaming.Duration(1000));

    SparkRunner.initAccumulators(options, jsc);

    TransformTranslator.Translator translator = new TransformTranslator.Translator();

    SparkNativePipelineVisitor visitor;
    if (isStreaming) {
      SparkPipelineTranslator streamingTranslator =
          new StreamingTransformTranslator.Translator(translator);
      EvaluationContext ctxt = new EvaluationContext(jsc, pipeline, options, jssc);
      visitor = new SparkNativePipelineVisitor(streamingTranslator, ctxt);
    } else {
      EvaluationContext ctxt = new EvaluationContext(jsc, pipeline, options, jssc);
      visitor = new SparkNativePipelineVisitor(translator, ctxt);
    }

    pipeline.traverseTopologically(visitor);

    SparkContextFactory.stopSparkContext(jsc);

    String debugString = visitor.getDebugString();
    LOG.info("Translated Native Spark pipeline:\n{}", debugString);
    return new DebugSparkPipelineResult(debugString);
  }
@Test
  public void debugStreamingPipeline() {
    PipelineOptions options = contextRule.configure(PipelineOptionsFactory.create());
    options.setRunner(SparkRunnerDebugger.class);
    options.as(TestSparkPipelineOptions.class).setStreaming(true);
    Pipeline pipeline = Pipeline.create(options);

    KafkaIO.Read<String, String> read =
        KafkaIO.<String, String>read()
            .withBootstrapServers("mykafka:9092")
            .withTopics(Collections.singletonList("my_input_topic"))
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class);

    KafkaIO.Write<String, String> write =
        KafkaIO.<String, String>write()
            .withBootstrapServers("myotherkafka:9092")
            .withTopic("my_output_topic")
            .withKeySerializer(StringSerializer.class)
            .withValueSerializer(StringSerializer.class);

    KvCoder<String, String> stringKvCoder = KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

    pipeline
        .apply(read.withoutMetadata())
        .setCoder(stringKvCoder)
        .apply(Window.into(FixedWindows.of(Duration.standardSeconds(5))))
        .apply(ParDo.of(new SparkRunnerDebuggerTest.FormatKVFn()))
        .apply(Distinct.create())
        .apply(WithKeys.of(new SparkRunnerDebuggerTest.ArbitraryKeyFunction()))
        .apply(write);

    final String expectedPipeline =
        "KafkaUtils.createDirectStream(...)\n"
            + "_.map(new org.apache.beam.sdk.transforms.windowing.FixedWindows())\n"
            + "_.mapPartitions(new org.apache.beam.runners.spark."
            + "SparkRunnerDebuggerTest$FormatKVFn())\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Distinct$1())\n"
            + "_.groupByKey()\n"
            + "_.map(new org.apache.beam.sdk.transforms.Combine$IterableCombineFn())\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Distinct$3())\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.WithKeys$1())\n"
            + "_.<org.apache.beam.sdk.io.kafka.AutoValue_KafkaIO_Write>";

    SparkRunnerDebugger.DebugSparkPipelineResult result =
        (SparkRunnerDebugger.DebugSparkPipelineResult) pipeline.run();

    assertThat(
        "Debug pipeline did not equal expected",
        result.getDebugString(),
        Matchers.equalTo(expectedPipeline));
  }