从流式Dataflow调用Google Dataflow RestAPI启动批处理时遇间歇性401未授权问题

看起来你的间歇性401问题根源在于访问令牌过期后没有自动刷新，咱们一步步拆解问题，然后给出更可靠的实现方案：

首先，分析你当前代码的问题：
你现在只在程序启动时获取一次AccessToken，但GCP的OAuth2访问令牌默认有效期是1小时。当运行30分钟后，令牌逐渐接近过期，后续请求就会返回401未授权错误。而且你直接提取了令牌的字符串值，没有利用ComputeEngineCredentials本身具备的自动刷新能力。

接下来是两种优化方案，优先推荐官方客户端库的方式，更省心：

方案1：使用Dataflow官方Java客户端库（推荐） #

官方库会自动处理凭据刷新、API调用的细节，不用自己手动构造HTTP请求和管理令牌。

首先添加Maven依赖（使用最新版本）：

<dependency>
  <groupId>com.google.cloud</groupId>
  <artifactId>google-cloud-dataflow</artifactId>
  <version>2.53.0</version> <!-- 可替换为最新版本号 -->
</dependency>

然后编写启动批处理作业的代码：

import com.google.cloud.dataflow.v1beta3.TemplatesClient;
import com.google.dataflow.v1beta3.LaunchTemplateParameters;
import com.google.dataflow.v1beta3.LaunchTemplateRequest;
import com.google.dataflow.v1beta3.LaunchTemplateResponse;

public class DataflowBatchLauncher {
    public void launchTemplate() throws Exception {
        // 自动使用默认应用凭据（Compute Engine服务账号），并自动刷新令牌
        try (TemplatesClient templatesClient = TemplatesClient.create()) {
            String projectId = "dataflow-begining";
            String templateGcsPath = "gs://template/test.json";

            // 可根据需求添加作业参数，比如worker数量、区域等
            LaunchTemplateParameters params = LaunchTemplateParameters.newBuilder()
                    .setRegion("us-central1") // 替换成你的Dataflow区域
                    .build();

            LaunchTemplateRequest request = LaunchTemplateRequest.newBuilder()
                    .setProjectId(projectId)
                    .setGcsPath(templateGcsPath)
                    .setLaunchParameters(params)
                    .build();

            LaunchTemplateResponse response = templatesClient.launchTemplate(request);
            System.out.println("成功启动批处理作业，Job ID: " + response.getJob().getId());
        }
    }
}

方案2：手动处理HTTP请求（如果必须用这种方式） #

如果你因为某些限制不能用官方库，那要确保每次请求前都获取有效的令牌，利用ComputeEngineCredentials的自动刷新机制：

import com.google.auth.oauth2.ComputeEngineCredentials;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;

public class ManualDataflowLauncher {
    public void launchBatchJob() throws Exception {
        // 获取默认凭据，这个实例会自动管理令牌的刷新
        ComputeEngineCredentials credentials = ComputeEngineCredentials.getApplicationDefault();

        // 检查令牌是否过期，提前1分钟刷新避免请求过程中令牌失效
        if (credentials.getAccessToken() == null 
            || credentials.getAccessToken().getExpirationTime().getTime() < System.currentTimeMillis() + 60000) {
            credentials.refresh();
        }

        String validToken = credentials.getAccessToken().getTokenValue();

        CloseableHttpClient client = HttpClientBuilder.create().useSystemProperties().build();
        HttpPost request = new HttpPost(
            "https://dataflow.googleapis.com/v1b3/projects/dataflow-begining/templates:launch?gcsPath=gs://template/test.json"
        );
        request.setHeader("Authorization", "Bearer " + validToken);

        // 执行请求并处理响应，可根据需求添加响应解析逻辑
        client.execute(request);
    }
}

关键注意点： #

1. 权限验证：确保运行代码的Compute Engine实例（或其他环境）的服务账号拥有dataflow.jobs.create权限，或者直接绑定roles/dataflow.admin角色，避免因权限不足导致的401（不过你的问题是间歇性的，这个可能性较低，但仍需确认）。
2. 令牌生命周期：GCP访问令牌有效期最长1小时，绝对不能缓存令牌并长期使用，必须依赖凭据实例的自动刷新能力。
3. 官方库优势：官方客户端库不仅处理令牌刷新，还会处理API的错误重试、参数验证等，比手动写HTTP请求更稳定可靠。

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