@Override
    public ConnectedComponentVertexProgram generateProgram(final Graph graph, final Memory memory) {
        final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
        detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));

        final ConnectedComponentVertexProgram.Builder builder = ConnectedComponentVertexProgram.build().
                edges(detachedTraversal).
                property(this.clusterProperty);

        if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
            final TraverserSet<?> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
            if (!haltedTraversers.isEmpty()) {
                Object haltedTraversersValue;
                try {
                    haltedTraversersValue = Base64.getEncoder().encodeToString(Serializer.serializeObject(haltedTraversers));
                } catch (final IOException ignored) {
                    haltedTraversersValue = haltedTraversers;
                }
                builder.configure(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversersValue);
            }
        }

        return builder.create(graph);
    }
@author Stephen Mallette (http://stephen.genoprime.com)
gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ConnectedComponentVertexProgramStep.java