@Override
  public DirectPipelineResult run(Pipeline pipeline) {
    try {
      options =
          MAPPER
              .readValue(MAPPER.writeValueAsBytes(options), PipelineOptions.class)
              .as(DirectOptions.class);
    } catch (IOException e) {
      throw new IllegalArgumentException(
          "PipelineOptions specified failed to serialize to JSON.", e);
    }

    performRewrites(pipeline);
    MetricsEnvironment.setMetricsSupported(true);
    try {
      DirectGraphVisitor graphVisitor = new DirectGraphVisitor();
      pipeline.traverseTopologically(graphVisitor);

      @SuppressWarnings("rawtypes")
      KeyedPValueTrackingVisitor keyedPValueVisitor = KeyedPValueTrackingVisitor.create();
      pipeline.traverseTopologically(keyedPValueVisitor);

      DisplayDataValidator.validatePipeline(pipeline);
      DisplayDataValidator.validateOptions(options);

      ExecutorService metricsPool =
          Executors.newCachedThreadPool(
              new ThreadFactoryBuilder()
                  .setThreadFactory(MoreExecutors.platformThreadFactory())
                  .setDaemon(false) // otherwise you say you want to leak, please don't!
                  .setNameFormat("direct-metrics-counter-committer")
                  .build());
      DirectGraph graph = graphVisitor.getGraph();
      EvaluationContext context =
          EvaluationContext.create(
              clockSupplier.get(),
              Enforcement.bundleFactoryFor(enabledEnforcements, graph),
              graph,
              keyedPValueVisitor.getKeyedPValues(),
              metricsPool);

      TransformEvaluatorRegistry registry =
          TransformEvaluatorRegistry.javaSdkNativeRegistry(context, options);
      PipelineExecutor executor =
          ExecutorServiceParallelExecutor.create(
              options.getTargetParallelism(),
              registry,
              Enforcement.defaultModelEnforcements(enabledEnforcements),
              context,
              metricsPool);
      executor.start(graph, RootProviderRegistry.javaNativeRegistry(context, options));

      DirectPipelineResult result = new DirectPipelineResult(executor, context);
      if (options.isBlockOnRun()) {
        try {
          result.waitUntilFinish();
        } catch (UserCodeException userException) {
          throw new PipelineExecutionException(userException.getCause());
        } catch (Throwable t) {
          if (t instanceof RuntimeException) {
            throw (RuntimeException) t;
          }
          throw new RuntimeException(t);
        }
      }
      return result;
    } finally {
      MetricsEnvironment.setMetricsSupported(false);
    }
  }
@Test
  public void testMutatingInputCoderDoFnError() throws Exception {
    Pipeline pipeline = getPipeline();

    pipeline
        .apply(Create.of(new byte[] {0x1, 0x2, 0x3}, new byte[] {0x4, 0x5, 0x6}))
        .apply(
            ParDo.of(
                new DoFn<byte[], Integer>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    byte[] inputArray = c.element();
                    inputArray[0] = 0xa;
                    c.output(13);
                  }
                }));

    thrown.expect(IllegalMutationException.class);
    thrown.expectMessage("Input");
    thrown.expectMessage("must not be mutated");
    pipeline.run();
  }