Flink Job拦截器处理429错误时，何时关闭Response？

你的Interceptor存在未关闭废弃Response资源的问题，直接在返回前执行response.body().close()是错误的——因为你返回的Response需要被后续Flink Job代码使用，关闭body后后续将无法读取响应内容。

核心问题出在哪？ #

当第一次请求返回429时，你直接用新的Response覆盖了变量，原来的429 Response就成了未处理的"孤儿"对象，它的body没有被关闭，导致底层连接/资源泄漏，这就是你收到警告的原因。

正确的修复代码 #

public Response intercept(Chain chain) throws IOException {
    Response response = chain.proceed(chain.request());

    if (!response.isSuccessful() && response.code() == 429) {
        LOG.error("Rate limit exceeded. Waiting 20 seconds.");
        // 关键：先关闭废弃的429响应的body，避免资源泄漏
        response.body().close();
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            // 不要吞掉中断，恢复线程中断状态并抛出异常
            Thread.currentThread().interrupt();
            throw new IOException("Interrupted while waiting for rate limit reset", e);
        }
        // 获取新的响应并覆盖变量
        response = chain.proceed(chain.request());
    }
    // 返回的响应不能在这里关闭body，后续调用方需要使用它
    return response;
}

关键注意事项 #

• 对于不需要传递给后续流程的Response（比如这里的429响应），必须手动关闭它的body，这是OkHttp的资源管理规范。
• 返回给上层的Response绝对不能提前关闭body，否则你的Flink Job在读取响应内容时会报错。
• 处理InterruptedException时，不要只打印堆栈，恢复线程的中断状态并抛出异常，这样上层逻辑能正确处理中断场景。

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