Понимание предупреждения Jetty Closed while Pending/Unready

Предупреждение Closed while Pending/Unready можно воспроизвести с помощью ручной отладки и простого клиента curl. Я тестировал его с Jetty 9.4.8.v20171121 и Jetty 9.4.11.v20180605. Вам нужны две точки останова, поэтому надежно воспроизвести в тестовой среде непросто.

1. Первая точка останова находится в методе HttpOutput.write() сразу после того, как он меняет свое состояние с READY на PENDING:

   case READY:
       if (!_state.compareAndSet(OutputState.READY, OutputState.PENDING))
           continue;
       // put a breakpoint here
   

2. Второй в Response.closeOutput():

   case STREAM:
       // put a breakpiont here
       if (!_out.isClosed())
           getOutputStream().close();
       break;
   

Действия по воспроизведению:

1. [JettyThread1] Вызывает QueueWriteListener.onWritePossible(), который записывает несколько байтов в вывод, а затем возвращается (поскольку его входной буфер пуст).
2. Дождитесь события onTimeout.
3. [JettyThread2] HttpChannelState.onTimeout() вызывает QueueWriteListener.onTimeout(), который вызывает asyncContext.complete().
4. [JettyThread2] HttpChannelState.onTimeout() планирует отправку по истечении асинхронного тайм-аута.
5. [JettyThread2] Пауза во второй точке останова
6. [ДПФ] DataFeederThread вызывает writeImpl()
7. [DFT] DataFeederThread вызывает HttpOutput.write() (это выходной поток)
8. [DFT] HttpOutput.write() меняет свое состояние с READY на PENDING
9. [DFT] DataFeederThread здесь останавливается из-за точки останова выше
10. [JettyThread2] Запланированная отправка закрывает выходной поток и выдает предупреждение Closed while Pending/Unready.

Итак, на самом деле именно Jetty закрывает выходной поток в этом (Jetty 9.4.8.v20171121) стеке:

Thread [jetty-19] (Suspended)   
    Response.closeOutput() line: 1043   
    HttpChannelOverHttp(HttpChannel).handle() line: 488 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748  

Создание onTimeout() synchronized (а также writeImpl() также synchronized) в слушателе не помогает, так как запланированное закрытие все еще может перекрываться с writeImpl (из DataFeederThread). Рассмотрим этот случай:

1. [JettyThread1] Вызывает QueueWriteListener.onWritePossible(), который записывает несколько байтов в вывод, а затем возвращается (поскольку его входной буфер пуст).
2. Дождитесь события onTimeout.
3. [JettyThread2] HttpChannelState.onTimeout() вызывает QueueWriteListener.onTimeout(), который вызывает asyncContext.complete().
4. [DFT] DataFeederThread вызывает writeImpl() (он заблокирован, так как onTimeout еще не закончил)
5. [JettyThread2] QueueWriteListener.onTimeout() завершено
6. [ДПФ] writeImpl() может работать
7. [JettyThread2] HttpChannelState.onTimeout() планирует отправку по истечении асинхронного тайм-аута.
8. [JettyThread2] Пауза во второй точке останова
9. [DFT] DataFeederThread вызывает HttpOutput.write() (это выходной поток)
10. [DFT] HttpOutput.write() меняет свое состояние с READY на PENDING
11. [DFT] DataFeederThread здесь останавливается из-за точки останова выше
12. [JettyThread2] Запланированная отправка закрывает выходной поток и выдает предупреждение Closed while Pending/Unready.

К сожалению, после asnyContext.complete() недостаточно проверить output.isReady(). Он возвращает true, так как Jetty повторно открывает HttpOutput (см. стек ниже), поэтому вам нужен отдельный флаг для этого в слушателе.

Thread [jetty-13] (Suspended (access of field _state in HttpOutput))    
    HttpOutput.reopen() line: 195   
    HttpOutput.recycle() line: 923  
    Response.recycle() line: 138    
    HttpChannelOverHttp(HttpChannel).recycle() line: 269    
    HttpChannelOverHttp.recycle() line: 83  
    HttpConnection.onCompleted() line: 424  
    HttpChannelOverHttp(HttpChannel).onCompleted() line: 695    
    HttpChannelOverHttp(HttpChannel).handle() line: 493 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748  

Кроме того, isReady() также возвращает true, когда выход все еще закрыт (перед перезапуском/повторным открытием). Связанный вопрос: isReady() возвращает true в закрытом состоянии - почему?

Окончательная реализация выглядит примерно так:

@Override
protected void doPost(final HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {
    resp.setContentType(MediaType.OCTET_STREAM.type());
    resp.setStatus(HttpServletResponse.SC_OK);
    resp.setBufferSize(4096);
    resp.flushBuffer();
    final AsyncContext async = req.startAsync();
    async.setTimeout(5_000); // millis
    final ServletOutputStream output = resp.getOutputStream();
    final QueueWriteListener writeListener = new QueueWriteListener(async, output);
    async.addListener(writeListener);
    output.setWriteListener(writeListener);
}

private static class QueueWriteListener implements AsyncListener, WriteListener {

    private static final Logger logger = LoggerFactory.getLogger(QueueWriteListener.class);

    private final AsyncContext asyncContext;
    private final ServletOutputStream output;

    @GuardedBy("this")
    private boolean completed = false;

    public QueueWriteListener(final AsyncContext asyncContext, final ServletOutputStream output) {
        this.asyncContext = checkNotNull(asyncContext, "asyncContext cannot be null");
        this.output = checkNotNull(output, "output cannot be null");
    }

    @Override
    public void onWritePossible() throws IOException {
        writeImpl();
    }

    private synchronized void writeImpl() throws IOException {
        if (completed) {
            return;
        }
        while (output.isReady()) {
            final byte[] message = getNextMessage();
            if (message == null) {
                output.flush();
                return;
            }
            output.write(message);
        }
    }

    private synchronized void completeImpl() {
        // also stops DataFeederThread to call bufferArrived
        completed = true;
        asyncContext.complete();
    }

    @Override
    public void onError(final Throwable t) {
        logger.error("Writer.onError", t);
        completeImpl();
    }

    public void dataArrived() {
        try {
            writeImpl();
        } catch (RuntimeException | IOException e) {
            ...
        }
    }

    public void noMoreData() {
        completeImpl();
    }

    @Override
    public synchronized void onComplete(final AsyncEvent event) throws IOException {
        completed = true; // might not needed but does not hurt
    }

    @Override
    public synchronized void onTimeout(final AsyncEvent event) throws IOException {
        completeImpl();
    }

    @Override
    public void onError(final AsyncEvent event) throws IOException {
        logger.error("onError", event.getThrowable());
    }

    ...
}

обновление 2018-08-01: на самом деле оно не устранило предупреждение полностью, см.: "Закрыто в ожидании/ Предупреждения о неготовности от Jetty

Вы используете асинхронный ввод-вывод из спецификации сервлета и вручную закрыли поток ответов.

Ключевой факт: вызов close подразумевает операцию записи.

В случае Jetty IOException сообщает вам, что ваш ручной вызов закрытия потока ответа вызвал нежелательный побочный эффект.

Таким образом, в режиме асинхронного ввода-вывода вызов ServletOutputStream.isReady() должно было быть выполнено до использования ServletOutputStream.close() для проверки истинности isReady().

Обратите внимание, что может быть желательно разрешить ServletOutputStream.close() в любое время, особенно если AsyncContext.complete().

Однако, если предыдущая запись еще не завершена и/или ServletOutputStream.isReady() не был вызван, этот конкретный случай голого ServletOutputStream.close() разрешен, но поведение состоит в том, чтобы прервать ответ, отбрасывая все ожидающие/незаписанные записи в ServletOutputStream.

Вот почему вы получаете IOException("Closed while Pending/Unready")

Теперь вы должны спросить себя, почему вы закрываете поток ответов? Это не требуется в соответствии со спецификацией сервлета и нежелательно, если вы когда-либо хотели использовать RequestDispatcher для объединения нескольких ответов сервлетов вместе.