@Override
    protected void sideEffect(final Traverser.Admin<S> traverser) {
        final TraversalSideEffects sideEffects = this.getTraversal().getSideEffects();
        // addAll and assign operator collect solutions as a BulkSet. And addAll merges them with an existing
        // sideEffect value which is expected to be Collection. assign replaces the existing value with the new BulkSet.
        // Therefore, they need a different treatment how they gather inputs.
        final boolean isOperatorForBulkSet = sideEffects.getReducer(sideEffectKey) == Operator.addAll ||
                sideEffects.getReducer(sideEffectKey) == Operator.assign;

        final BulkSet<Object> bulkSet = new BulkSet<>();
        TraversalUtil.produce(traverser, this.storeTraversal).ifProductive(p -> bulkSet.add(p, traverser.bulk()));

        if (isOperatorForBulkSet) {
            sideEffects.add(this.sideEffectKey, bulkSet);
        } else {
            bulkSet.forEach(p -> sideEffects.add(sideEffectKey, p));
        }
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateLocalStep.java