private void iterateScriptEvalResult(final Context context, MessageSerializer<?> serializer, final RequestMessage message)
            throws ProcessingException, InterruptedException, ScriptException {
        if (message.optionalField(Tokens.ARGS_BINDINGS).isPresent()) {
            final Map bindings = (Map) message.getFields().get(Tokens.ARGS_BINDINGS);
            if (IteratorUtils.anyMatch(bindings.keySet().iterator(), k -> null == k || !(k instanceof String))) {
                throw new ProcessingException(GremlinError.binding());
            }

            final Set<String> badBindings = IteratorUtils.set(IteratorUtils.<String>filter(bindings.keySet().iterator(), INVALID_BINDINGS_KEYS::contains));
            if (!badBindings.isEmpty()) {
                throw new ProcessingException(GremlinError.binding(badBindings));
            }

            // ignore control bindings that get passed in with the "#jsr223" prefix - those aren't used in compilation
            if (IteratorUtils.count(IteratorUtils.filter(bindings.keySet().iterator(), k -> !k.toString().startsWith("#jsr223"))) > settings.maxParameters) {
                throw new ProcessingException(GremlinError.binding(bindings.size(), settings.maxParameters));
            }
        }

        final Map<String, Object> args = message.getFields();
        final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : "gremlin-lang";
        final GremlinScriptEngine scriptEngine = gremlinExecutor.getScriptEngineManager().getEngineByName(language);

        final Bindings mergedBindings = mergeBindingsFromRequest(context, new SimpleBindings(graphManager.getAsBindings()));
        final Object result = scriptEngine.eval(message.getGremlin(), mergedBindings);

        final String bulkingSetting = context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULK_RESULTS);
        // bulking only applies if it's gremlin-lang, and per request token setting takes precedence over header setting.
        // The serializer check is temporarily needed because GraphSON hasn't been removed yet and doesn't support bulking.
        final boolean bulking = language.equals("gremlin-lang") && serializer instanceof GraphBinaryMessageSerializerV4 ?
                (args.containsKey(Tokens.BULK_RESULTS) ?
                        Objects.equals(args.get(Tokens.BULK_RESULTS), "true") :
                        Objects.equals(bulkingSetting, "true")) :
                false;

        Iterator itty = null;
        try {
            if (bulking) {
                // optimization for driver requests
                ((Traversal.Admin<?, ?>) result).applyStrategies();
                itty = new TraverserIterator((Traversal.Admin<?, ?>) result);
                handleIterator(context, itty, serializer, true);
            } else {
                itty = IteratorUtils.asIterator(result);
                handleIterator(context, itty, serializer, false);
            }
        } catch (Exception ex) {
            // TINKERPOP-3144 ensure Traversals are closed when exception thrown.
            if (itty instanceof TraverserIterator) {
                CloseableIterator.closeIterator(((TraverserIterator) itty).getTraversal());
            } else if (itty != null) {
                CloseableIterator.closeIterator(itty);
            }

            throw ex;
        }
    }
Tracks the rate of pause to writes when the high watermark is exceeded.
gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java