使用Google BigQuery Data Transfer实现S3迁移时如何等待传输完成？

BigQuery Data Transfer官方Java SDK完全支持查询传输任务的运行状态，第五步可以通过轮询任务状态的轻量方案实现，不需要额外对接其他服务。

核心实现步骤 #

• 调用startManualTransferRuns()触发任务后，从返回的StartManualTransferRunsResponse中提取本次任务对应的TransferRun实例，拿到唯一资源标识runName
• 启动定时轮询逻辑，建议间隔设置为10~30秒（数据量越大可以适当拉长间隔，避免触发API限流），每次调用getTransferRun()接口传入runName拉取最新任务状态
• 当任务进入终态后终止轮询：状态为SUCCEEDED时执行后续业务逻辑，状态为FAILED/CANCELLED时走异常处理分支

示例代码 #

import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.GetTransferRunRequest;
import com.google.cloud.bigquery.datatransfer.v1.StartManualTransferRunsResponse;
import com.google.cloud.bigquery.datatransfer.v1.TransferRun;
import java.util.concurrent.TimeUnit;

// 步骤1~4的初始化逻辑省略，直接从拿到手动触发响应开始处理
public void waitForTransferFinish(StartManualTransferRunsResponse triggerResponse, DataTransferServiceClient client) throws InterruptedException {
    // 取出本次触发的传输任务实例
    TransferRun currentRun = triggerResponse.getRuns(0);
    String runResourceName = currentRun.getName();
    TransferRun.State runState = currentRun.getState();

    // 配置最大等待时间，避免无限轮询 示例为2小时
    long maxWaitTime = TimeUnit.HOURS.toMillis(2);
    long startTime = System.currentTimeMillis();

    // 轮询直到任务进入终态或超时
    while (!isRunEnded(runState)) {
        if (System.currentTimeMillis() - startTime > maxWaitTime) {
            throw new RuntimeException("传输任务超时，当前状态：" + runState.name());
        }
        // 轮询间隔15秒，可根据业务调整
        TimeUnit.SECONDS.sleep(15);
        // 拉取最新任务状态
        GetTransferRunRequest getRunRequest = GetTransferRunRequest.newBuilder()
                .setName(runResourceName)
                .build();
        currentRun = client.getTransferRun(getRunRequest);
        runState = currentRun.getState();
    }

    // 终态处理
    if (runState == TransferRun.State.SUCCEEDED) {
        // 传输成功，执行后续业务逻辑
        System.out.println("S3到BigQuery传输完成，开始执行后续操作");
    } else {
        throw new RuntimeException("传输任务失败，最终状态：" + runState.name() + "，错误信息：" + currentRun.getErrorMessage());
    }
}

// 判断任务是否进入终态
private boolean isRunEnded(TransferRun.State state) {
    return state == TransferRun.State.SUCCEEDED
            || state == TransferRun.State.FAILED
            || state == TransferRun.State.CANCELLED;
}

可选优化方案 #

如果不想自行维护轮询逻辑，可以给TransferConfig配置Google Cloud Pub/Sub状态通知，任务状态变更时会自动推送事件到指定主题，直接消费事件即可触发后续操作。

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