Helidon SE中MessageBodyFilter使用DataChunkInputStream触发IOException

问题原因 #

你用try-with-resources创建DataChunkInputStream时，try代码块执行完毕就会自动关闭流，但supplyAsync里的异步任务是在try块结束后才执行的，这时候流已经被关闭，读取自然会抛出IOException。不用try-with-resources的话，流没有被提前关闭，所以能正常读取。

解决办法 #

把流的创建和读取逻辑放到异步任务内部，手动控制流的关闭时机，确保流在读取完成后再释放资源。

修改后的代码示例：

@Override
public Flow.Publisher<DataChunk> apply(final Flow.Publisher<DataChunk> dataChunkPublisher) {
    final var span = tracer.buildSpan("request-preprocessing")
            .asChildOf(spanContext)
            .start();
    return Single.create(CompletableFuture.supplyAsync(() -> Contexts.runInContext(ctx, () -> {
        InputStream dataChunkInputStream = null;
        try {
            dataChunkInputStream = new DataChunkInputStream(dataChunkPublisher);
            final var jsonNode = OM.readTree(dataChunkInputStream);
            // 这里添加你的负载字段修改逻辑
            // ...
            // 修改完成后将JsonNode转回DataChunk并返回
            byte[] modifiedBytes = OM.writeValueAsBytes(jsonNode);
            DataChunk chunk = DataChunk.create(true, modifiedBytes);
            return chunk;
        } catch (IOException e) {
            span.recordException(e);
            throw new RuntimeException("处理请求负载失败", e);
        } finally {
            if (dataChunkInputStream != null) {
                try {
                    dataChunkInputStream.close();
                } catch (IOException e) {
                    span.recordException(e);
                }
            }
            span.end();
        }
    })));
}

关键说明 #

• 将DataChunkInputStream的创建移到异步任务内部，让流的生命周期和异步任务绑定，避免被提前关闭。
• 在finally块手动关闭流，确保资源正常释放，避免内存泄漏。
• 修改完负载后，要将JsonNode转回DataChunk格式返回，保证后续流程能拿到修改后的请求内容。

内容的提问来源于stack exchange，提问作者Casey Daly