class Traverser extends AlgoTraverser {

        public Traverser(UserJob<Object> job) {
            super(job);
        }

        protected GraphTraversal<Vertex, Vertex> constructSource(
                String sourceLabel,
                long sourceSample,
                String sourceCLabel) {
            GraphTraversal<Vertex, Vertex> t = this.graph().traversal()
                                                   .withSack(1f).V();

            if (sourceLabel != null) {
                t = t.hasLabel(sourceLabel);
            }

            t = t.filter(it -> {
                this.updateProgress(++this.progress);
                return sourceCLabel == null || match(it.get(), sourceCLabel);
            });

            if (sourceSample > 0L) {
                t = t.sample((int) sourceSample);
            }

            return t;
        }

        protected GraphTraversal<Vertex, Vertex> constructPath(
                GraphTraversal<Vertex, Vertex> t, Directions dir,
                String label, long degree, long sample,
                String sourceLabel, String sourceCLabel) {
            GraphTraversal<?, Vertex> unit = constructPathUnit(dir, label,
                                                               degree, sample,
                                                               sourceLabel,
                                                               sourceCLabel);
            t = t.as("v").repeat(__.local(unit).simplePath().as("v"));

            return t;
        }

        protected GraphTraversal<Vertex, Vertex> constructPathUnit(
                Directions dir, String label,
                long degree, long sample,
                String sourceLabel,
                String sourceCLabel) {
            if (dir == null) {
                dir = Directions.BOTH;
            }
            Direction direction = dir.direction();

            String[] labels = {};
            if (label != null) {
                labels = new String[]{label};
            }

            GraphTraversal<Vertex, Vertex> unit = __.to(direction, labels);
            if (sourceLabel != null) {
                unit = unit.hasLabel(sourceLabel);
            }
            if (sourceCLabel != null) {
                unit = unit.has(C_LABEL, sourceCLabel);
            }
            if (degree != NO_LIMIT) {
                unit = unit.limit(degree);
            }
            if (sample > 0L) {
                unit = unit.sample((int) sample);
            }
            return unit;
        }

        protected <V> GraphTraversal<V, V> filterNonShortestPath(
                GraphTraversal<V, V> t,
                boolean keepOneShortestPath) {
            long size = this.graph().traversal().V().limit(100000L)
                            .count().next();
            Map<Pair<Id, Id>, Integer> triples = new HashMap<>((int) size);
            return t.filter(it -> {
                Id start = it.<HugeElement>path(Pop.first, "v").id();
                Id end = it.<HugeElement>path(Pop.last, "v").id();
                int len = it.path().size();
                assert len == it.<List<?>>path(Pop.all, "v").size();

                Pair<Id, Id> key = Pair.of(start, end);
                Integer shortest = triples.get(key);
                if (shortest != null && len > shortest) {
                    // ignore non shortest path
                    return false;
                }
                // TODO: len may be smaller than shortest
                if (shortest == null) {
                    triples.put(key, len);
                } else {
                    assert len == shortest;
                    return !keepOneShortestPath;
                }
                return true;
            });
        }

        protected GraphTraversal<Vertex, Id> substractPath(
                GraphTraversal<Vertex, Vertex> t,
                boolean withBoundary) {<FILL_FUNCTION_BODY>}

        protected GraphTraversal<Vertex, ?> topN(GraphTraversal<Vertex, ?> t,
                                                 long topN) {
            if (topN > 0L || topN == NO_LIMIT) {
                t = t.order(Scope.local).by(Column.values, Order.desc);
                if (topN > 0L) {
                    assert topN != NO_LIMIT;
                    t = t.limit(Scope.local, topN);
                }
            }
            return t;
        }
    }

            // t.select(Pop.all, "v").unfold().id()
            return t.select(Pop.all, "v").flatMap(it -> {
                List<?> path = (List<?>) it.get();
                if (withBoundary) {
                    @SuppressWarnings("unchecked")
                    Iterator<HugeVertex> items = (Iterator<HugeVertex>)
                            path.iterator();
                    return new MapperIterator<>(items, HugeVertex::id);
                }
                int len = path.size();
                if (len < 3) {
                    return Collections.emptyIterator();
                }

                LOG.debug("CentAlgorithm substract path: {}", path);
                path.remove(path.size() - 1);
                path.remove(0);
                @SuppressWarnings("unchecked")
                Iterator<HugeVertex> items = (Iterator<HugeVertex>)
                        path.iterator();
                return new MapperIterator<>(items, HugeVertex::id);
            });
        
<methods>public non-sealed void <init>() ,public void checkParameters(Map<java.lang.String,java.lang.Object>) <variables>public static final int BATCH,public static final java.lang.String CATEGORY_AGGR,public static final java.lang.String CATEGORY_CENT,public static final java.lang.String CATEGORY_COMM,public static final java.lang.String CATEGORY_PATH,public static final java.lang.String CATEGORY_RANK,public static final java.lang.String CATEGORY_SIMI,public static final java.lang.String C_LABEL,public static final double DEFAULT_ALPHA,public static final long DEFAULT_CAPACITY,public static final long DEFAULT_DEGREE,public static final long DEFAULT_EACH_LIMIT,public static final long DEFAULT_LIMIT,public static final double DEFAULT_PRECISION,public static final long DEFAULT_SAMPLE,public static final long DEFAULT_STABLE_TIMES,public static final long DEFAULT_TIMES,public static final java.lang.String EXPORT_PATH,public static final java.lang.String KEY_ALPHA,public static final java.lang.String KEY_CAPACITY,public static final java.lang.String KEY_CLEAR,public static final java.lang.String KEY_DEGREE,public static final java.lang.String KEY_DEPTH,public static final java.lang.String KEY_DIRECTION,public static final java.lang.String KEY_EACH_LIMIT,public static final java.lang.String KEY_EXPORT_COMM,public static final java.lang.String KEY_LABEL,public static final java.lang.String KEY_LIMIT,public static final java.lang.String KEY_PRECISION,public static final java.lang.String KEY_SAMPLE,public static final java.lang.String KEY_SHOW_COMM,public static final java.lang.String KEY_SHOW_MOD,public static final java.lang.String KEY_SKIP_ISOLATED,public static final java.lang.String KEY_SOURCE_CLABEL,public static final java.lang.String KEY_SOURCE_LABEL,public static final java.lang.String KEY_SOURCE_SAMPLE,public static final java.lang.String KEY_STABLE_TIMES,public static final java.lang.String KEY_TIMES,public static final java.lang.String KEY_TOP,public static final java.lang.String KEY_WORKERS,public static final long MAX_CAPACITY,public static final long MAX_QUERY_LIMIT,public static final long MAX_RESULT_SIZE,public static final java.lang.String R_RANK,public static final java.lang.String USER_DIR