@Override
    public Map<K, V> projectTraverser(final Traverser.Admin<S> traverser) {
        final Map<K, V> map = new HashMap<>(1);
        this.valueTraversal.reset();
        this.valueTraversal.addStart(traverser);

        TraversalUtil.produce(traverser, this.keyTraversal).ifProductive(p -> {
            // if there's no barrier then add the result of the valueTraversal as the value. if there is a barrier,
            // then pop the next barrier value into the map. there are cases where hasNextBarrier is false in which
            // case CollectingBarrierStep instances get special handling but only when there are
            // Supplying/ReducingBarrierStep somewhere in the traversal. these step instances like fold() or sum()
            // can take an empty TraverserSet and actually produce an output. if that isn't accounted for the key will
            // just get filtered out. TINKERPOP-2971
            if (null == this.barrierStep) {
                if (this.valueTraversal.hasNext()) {
                    map.put((K) p, (V) this.valueTraversal.next());
                }
            } else if (this.barrierStep.hasNextBarrier()) {
                map.put((K) p, (V) this.barrierStep.nextBarrier());
            } else if (this.barrierStep instanceof FilteringBarrier) {
                final int barrierStepIndex = TraversalHelper.stepIndex((Step) barrierStep, valueTraversal) + 1;
                if (valueTraversal.getSteps().stream().skip(barrierStepIndex).anyMatch(
                            step -> step instanceof SupplyingBarrierStep || step instanceof ReducingBarrierStep)) {
                    map.put((K) p, (V) this.barrierStep.getEmptyBarrier());
                }
            }
        });

        return map;
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java