基于Apache Beam读取带前缀的Cloud Pub/Sub多订阅的ValueProvider问题

这确实是Beam模板化场景里非常常见的痛点——因为expand()方法是在模板编译阶段（而非作业运行阶段）执行的，而ValueProvider的实际值要到作业启动时才会被解析，所以直接在expand()里用前缀去匹配订阅肯定行不通。下面给你两种可行的解决方案，你可以根据自己的场景选择：

方案1：运行时动态获取符合前缀的订阅列表（推荐用于订阅频繁变化的场景） #

核心思路是实现一个自定义的SubscriptionProvider，让它在作业运行时通过Pub/Sub Admin API读取符合前缀的订阅列表，再传递给PubsubIO。

步骤1：实现自定义SubscriptionProvider #

这个类会在运行时解析ValueProvider的实际前缀值，然后调用API列出匹配的订阅：

import com.google.cloud.pubsub.v1.PubSubAdminClient;
import com.google.pubsub.v1.ProjectName;
import com.google.pubsub.v1.Subscription;
import org.apache.beam.sdk.io.gcp.pubsub.SubscriptionProvider;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

public class PrefixSubscriptionProvider implements SubscriptionProvider {
    private final ValueProvider<String> subscriptionPrefix;
    private final String projectId;

    public PrefixSubscriptionProvider(ValueProvider<String> subscriptionPrefix, String projectId) {
        this.subscriptionPrefix = subscriptionPrefix;
        this.projectId = projectId;
    }

    @Override
    public List<String> getSubscriptions() throws IOException {
        // 运行时才获取前缀的实际值
        String prefix = subscriptionPrefix.get();
        try (PubSubAdminClient adminClient = PubSubAdminClient.create()) {
            // 列出当前项目下所有符合前缀的订阅
            return adminClient.listSubscriptions(ProjectName.of(projectId))
                    .iterateAll()
                    .stream()
                    .map(Subscription::getName)
                    .filter(subName -> subName.startsWith(prefix))
                    .collect(Collectors.toList());
        }
    }
}

步骤2：在自定义PTransform中使用这个Provider #

将你的PTransform修改为使用这个动态Provider，这样expand()阶段只会定义读取逻辑，实际的订阅列表会在运行时获取：

import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.options.ValueProvider;

public class MultiPubsubReader extends PTransform<PBegin, PCollection<PubsubMessage>> {
    private final ValueProvider<String> subscriptionPrefix;
    private final String projectId;

    public MultiPubsubReader(ValueProvider<String> subscriptionPrefix, String projectId) {
        this.subscriptionPrefix = subscriptionPrefix;
        this.projectId = projectId;
    }

    @Override
    public PCollection<PubsubMessage> expand(PBegin input) {
        SubscriptionProvider provider = new PrefixSubscriptionProvider(subscriptionPrefix, projectId);
        return input.apply(PubsubIO.readMessagesWithAttributes()
                .fromSubscriptionProvider(provider));
    }
}

注意事项 #

• 权限配置：作业运行的服务账号需要拥有roles/pubsub.viewer权限（用于列出订阅）和roles/pubsub.subscriber权限（用于读取订阅消息）。
• 模板验证：模板创建阶段（dry run）无法验证订阅存在，所以要确保作业启动时前缀对应的订阅已经存在，或者在代码中添加异常处理逻辑。

方案2：预先传入订阅列表（适合订阅相对固定的场景） #

如果你的订阅不会频繁变化，可以提前通过外部工具（比如Terraform、Cloud Function）生成符合前缀的订阅列表，然后将这个列表作为ValueProvider<List<String>>传入模板，直接用PubsubIO的批量读取能力：

import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.options.ValueProvider;

import java.util.List;

public class MultiPubsubReader extends PTransform<PBegin, PCollection<PubsubMessage>> {
    private final ValueProvider<List<String>> subscriptions;

    public MultiPubsubReader(ValueProvider<List<String>> subscriptions) {
        this.subscriptions = subscriptions;
    }

    @Override
    public PCollection<PubsubMessage> expand(PBegin input) {
        return input.apply(PubsubIO.readMessagesWithAttributes()
                .fromSubscriptions(subscriptions));
    }
}

这种方式更简单，不需要调用Admin API，也避免了权限上的额外配置，唯一的缺点是需要外部维护订阅列表和前缀的同步。

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