@Override
    public void apply(final Traversal.Admin<?, ?> traversal) {
        final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // best guess at what the graph will be as its dynamically determined
        for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) {
            final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, EmptyMemory.instance()).getTraversal().get().clone();
            if (!computerTraversal.isLocked())
                computerTraversal.applyStrategies();
            if (SparkStarBarrierInterceptor.isLegal(computerTraversal)) {
                step.setComputer(step.getComputer()
                        .configure(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, true)
                        .configure(Constants.GREMLIN_SPARK_SKIP_GRAPH_CACHE, true)
                        .configure(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, SparkStarBarrierInterceptor.class.getCanonicalName()));
            }
        }
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategy.java