如何配置Flink RestClient发送Basic Auth头提交作业？

问题背景 #

官方Flink文档说明，若要保护Job Manager REST API，需在其前端部署反向代理（如Nginx）并配置认证，但这并非当前问题所在。当前问题是如何让flink-runtime库中的Flink RestClient发送BASIC Auth请求头，经代码分析发现该客户端不允许自定义作业提交请求。

可行解决方案 #

1. 通过RestClientConfiguration注入自定义头（推荐） #

从Flink 1.15版本开始，RestClientConfiguration支持直接设置自定义请求头。你可以在构建RestClient实例时，将Basic Auth的认证头传入配置：

// 生成Basic Auth头
String basicAuth = "Basic " + Base64.getEncoder().encodeToString("your-username:your-password".getBytes(StandardCharsets.UTF_8));

// 构建RestClient配置
RestClientConfiguration config = RestClientConfiguration.builder()
    .setAddress("your-jobmanager-host")
    .setPort(8081)
    .setHeaders(Collections.singletonMap("Authorization", basicAuth))
    .build();

// 创建带认证的RestClient
RestClient restClient = new RestClient(config, Executors.newCachedThreadPool());

这种方式是官方支持的方案，兼容性最好，不需要依赖内部API。如果你的Flink版本低于1.15，建议优先升级版本。

2. 反射修改RestClient内部HttpClient（兼容旧版本） #

如果无法升级Flink版本，可以通过反射获取RestClient内部的HttpClient实例，添加请求拦截器自动注入Auth头：

// 假设已有初始化好的RestClient实例
RestClient restClient = ...;

// 通过反射获取内部的HttpClient
Field httpClientField = RestClient.class.getDeclaredField("httpClient");
httpClientField.setAccessible(true);
CloseableHttpClient originalClient = (CloseableHttpClient) httpClientField.get(restClient);

// 构建带拦截器的新HttpClient
CloseableHttpClient authClient = HttpClients.custom()
    .setHttpClient(originalClient)
    .addInterceptorFirst((HttpRequest request, HttpContext context) -> {
        String basicAuth = "Basic " + Base64.getEncoder().encodeToString("your-username:your-password".getBytes(StandardCharsets.UTF_8));
        request.addHeader("Authorization", basicAuth);
    })
    .build();

// 替换RestClient中的原HttpClient
httpClientField.set(restClient, authClient);

注意：这种方法依赖Flink内部实现细节，版本更新可能导致字段名或结构变化，需要做好测试和兼容性处理。

3. 扩展CliFrontend提交作业 #

如果是通过代码模拟CLI提交作业，可以扩展CliFrontend类，在其构建RestClient的逻辑中注入Auth头，避免直接操作RestClient实例。

为什么不建议自行实现HTTP客户端 #

自行编写HTTP客户端调用JobManager REST API需要处理大量细节：比如请求体的序列化、响应的解析、状态码处理、版本兼容性等。Flink的REST API会随着版本迭代发生变化，自行维护的成本极高，优先利用官方提供的RestClient扩展方案更可靠。

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