Apache Beam使用PubsubIO.write写入Pubsub的失败处理方案咨询

核心原因说明 #

你直接在PTransform定义处捕获异常不生效，是因为PubsubIO的写入逻辑是运行在工作节点的异步执行链路中，异常并非在你定义expand方法的构造阶段抛出，外层的try-catch无法捕获到工作节点运行时抛出的异常。

可行方案 #

方案1：使用PubsubIO原生死信队列+重试配置（推荐） #

Apache Beam 2.29及以上版本的PubsubIO已经原生支持写入重试和死信输出能力，无需自己实现异常捕获逻辑：

• 可配置可重试异常类型，你遇到的UnavailableException本身就在默认可重试异常列表中
• 可自定义重试次数、重试间隔，满足你在连接恢复前持续重试、管道不挂的需求
• 超过重试次数仍失败的消息会自动写入你指定的死信主题，方便后续排查定位失败数据

代码示例：

import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
import com.google.common.base.Charsets;
import com.google.common.collect.ImmutableMap;
import org.joda.time.Duration;

public class PubSubWrite extends PTransform<PCollection<String>, PCollection<PubsubMessage>> {

    private final String outputTopic;
    private final String deadLetterTopic;

    public PubSubWrite(String outputTopic, String deadLetterTopic) {
        this.outputTopic = outputTopic;
        this.deadLetterTopic = deadLetterTopic;
    }

    @Override
    public PCollection<PubsubMessage> expand(PCollection<String> input) {
        return input
                .apply(
                        "convertMessagesToPubsubMessages",
                        MapElements.into(TypeDescriptor.of(PubsubMessage.class))
                                .via(
                                        (String json) ->
                                                new PubsubMessage(json.getBytes(Charsets.UTF_8), ImmutableMap.of("SOURCE", "TEST"))))
                .apply(
                        "writePubsubMessagesToPubSub", 
                        PubsubIO.writeMessages()
                                .to(outputTopic)
                                // 配置重试策略：最大重试次数10次，初始重试间隔1s，指数退避
                                .withRetryConfiguration(
                                        PubsubIO.RetryConfiguration.create(
                                                10, // 可根据需求调整为更大值，甚至Integer.MAX_VALUE实现无限重试
                                                Duration.standardSeconds(1)
                                        )
                                )
                                // 配置死信主题，超过重试次数的消息会写入这里
                                .withDeadLetterTopic(deadLetterTopic)
                );
    }
}

如果你不需要死信输出，只希望无限重试保证管道不挂，可以把重试次数设置为Integer.MAX_VALUE，这样会一直重试直到连接恢复。

方案2：自定义DoFn实现写入逻辑（适合自定义需求场景） #

如果你需要更灵活的失败处理逻辑（比如把失败消息写入其他存储、自定义日志埋点定位失败数据），可以自己封装DoFn实现Pubsub写入，主动捕获异常：

import com.google.api.gax.rpc.UnavailableException;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PubsubMessage;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.TupleTag;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class CustomPubsubWriteFn extends DoFn<String, Void> {
    private final String outputTopic;
    private Publisher publisher;
    public static final TupleTag<String> FAILED_MESSAGES_TAG = new TupleTag<String>(){};

    public CustomPubsubWriteFn(String outputTopic) {
        this.outputTopic = outputTopic;
    }

    @Setup
    public void setup() throws IOException {
        ProjectTopicName topicName = ProjectTopicName.parse(outputTopic);
        publisher = Publisher.newBuilder(topicName).build();
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        String message = c.element();
        int retryCount = 0;
        boolean success = false;
        while (retryCount < Integer.MAX_VALUE) { // 可自定义重试上限
            try {
                PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
                        .setData(ByteString.copyFromUtf8(message))
                        .putAttributes("SOURCE", "TEST")
                        .build();
                publisher.publish(pubsubMessage).get();
                success = true;
                break;
            } catch (UnavailableException e) {
                // 捕获连接异常，等待后重试
                retryCount++;
                try {
                    TimeUnit.SECONDS.sleep(Math.min(30, (long) Math.pow(2, retryCount))); // 指数退避，最大30s间隔
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            } catch (Exception e) {
                // 其他不可重试异常，输出到侧输出流
                break;
            }
        }
        if (!success) {
            c.output(FAILED_MESSAGES_TAG, message);
        }
    }

    @Teardown
    public void teardown() throws InterruptedException {
        if (publisher != null) {
            publisher.shutdown();
            publisher.awaitTermination(1, TimeUnit.MINUTES);
        }
    }
}

这种方式你可以完全控制异常处理逻辑，所有写入失败的消息都可以从侧输出流拿到，方便定位问题。

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