@Override
    public void channelRead0(final ChannelHandlerContext ctx, final RequestMessage requestMessage) {
        ctx.channel().attr(StateKey.HTTP_RESPONSE_SENT).set(false);
        final Pair<String, MessageSerializer<?>> serializer = ctx.channel().attr(StateKey.SERIALIZER).get();

        final Context requestCtx = new Context(requestMessage, ctx, settings, graphManager, gremlinExecutor,
                gremlinExecutor.getScheduledExecutorService(), NOT_STARTED);

        final Timer.Context timerContext = evalOpTimer.time();
        // timeout override - handle both deprecated and newly named configuration. earlier logic should prevent
        // both configurations from being submitted at the same time
        final Long timeoutMs = requestMessage.getField(Tokens.TIMEOUT_MS);
        final long seto = (null != timeoutMs) ? timeoutMs : requestCtx.getSettings().getEvaluationTimeout();

        final FutureTask<Void> evalFuture = new FutureTask<>(() -> {
            requestCtx.setStartedResponse();

            try {
                logger.debug("Processing request containing script [{}] and bindings of [{}] on {}",
                        requestMessage.getFieldOrDefault(Tokens.ARGS_GREMLIN, ""),
                        requestMessage.getFieldOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()),
                        Thread.currentThread().getName());
                if (settings.enableAuditLog) {
                    AuthenticatedUser user = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get();
                    if (null == user) {    // This is expected when using the AllowAllAuthenticator
                        user = AuthenticatedUser.ANONYMOUS_USER;
                    }
                    String address = ctx.channel().remoteAddress().toString();
                    if (address.startsWith("/") && address.length() > 1) address = address.substring(1);
                    auditLogger.info("User {} with address {} requested: {}", user.getName(), address,
                            requestMessage.getGremlin());
                }

                // Send back the 200 OK response header here since the response is always chunk transfer encoded. Any
                // failures that follow this will show up in the response body instead.
                final HttpResponse responseHeader = new DefaultHttpResponse(HTTP_1_1, OK);
                if (acceptsDeflateEncoding(ctx.attr(StateKey.REQUEST_HEADERS).get().getAll(ACCEPT_ENCODING))) {
                    responseHeader.headers().add(CONTENT_ENCODING, DEFLATE);
                }
                responseHeader.headers().set(TRANSFER_ENCODING, CHUNKED);
                responseHeader.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0());
                ctx.writeAndFlush(responseHeader);
                ctx.channel().attr(StateKey.HTTP_RESPONSE_SENT).set(true);

                iterateScriptEvalResult(requestCtx, serializer.getValue1(), requestMessage);
            } catch (Throwable t) {
                writeError(requestCtx, formErrorResponseMessage(t, requestMessage), serializer.getValue1());
            } finally {
                timerContext.stop();

                // There is a race condition that this query may have finished before the timeoutFuture was created,
                // though this is very unlikely. This is handled in the settor, if this has already been grabbed.
                // If we passed this point and the setter hasn't been called, it will cancel the timeoutFuture inside
                // the setter to compensate.
                final ScheduledFuture<?> timeoutFuture = requestCtx.getTimeoutExecutor();
                if (null != timeoutFuture)
                    timeoutFuture.cancel(true);
            }

            return null;
        });

        try {
            final Future<?> executionFuture = requestCtx.getGremlinExecutor().getExecutorService().submit(evalFuture);
            if (seto > 0) {
                // Schedule a timeout in the thread pool for future execution
                requestCtx.setTimeoutExecutor(requestCtx.getScheduledExecutorService().schedule(() -> {
                    executionFuture.cancel(true);
                    if (!requestCtx.getStartedResponse()) {
                        writeError(requestCtx, GremlinError.timeout(requestMessage), serializer.getValue1());
                    }
                }, seto, TimeUnit.MILLISECONDS));
            }
        } catch (RejectedExecutionException ree) {
            writeError(requestCtx, GremlinError.rateLimiting(), serializer.getValue1());
        }
    }
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