问题标签 [google-cloud-pubsublite]

发布者/订阅者是否可能位于与 Lite 安装位置不同的区域？例如两个用例 1. 发布者在欧洲的 GKE 应用程序中，但 Lite 及其订阅者在美国
2. 发布者和 Lite 在同一区域，例如欧洲，但订阅者在美国

首先，我想将这篇文章标记为 google-cloud-pubsub-lite，但它还没有创建，我很抱歉

我正在尝试使用 pubsub lite 进行介绍。我认为它可以用作在 GCP 项目中获取事件存储的“廉价”方式。

我们通常创建 GAE 标准服务，因此我们为使用的服务付费，同时它提供了出色的可扩展性。

阅读有关当前如何订阅 pubsub lite 的示例，我发现没有提供端点来接收新消息的选项。客户端连接到订阅并等待新消息流式传输抛出连接。

我想知道几个问题：

• 我们可以在 Cloud Function 或 GAE 标准服务的端点中接收来自 pubsub lite 主题的消息吗？
• 我们如何扩展到多个客户的主题订阅

谢谢

Description

• I have a pubSubSource connector in Kafka Connect Distributed mode that is simply reading from a PubSub subscription and writing into a Kafka topic. The issue is, even if I am publishing one message to GCP PubSub, I am getting this message written twice in my Kafka topic.

How to reproduce

• Deploy Kafka and Kafka connect

• Create a connector with below pubSubSource configurations:

• Below are the Kafka-connect configurations:

• Publish a message to the PubSub topic using the below command:

• Read messages from the destination Kafka topics:

Why is this happening, is there something that I am doing wrong?

当评论回复更新时，我能够收到推送通知。但是对于新问题，更新问题和问题的答案不会通过推送通知通知。我订阅了所有更改。

网址：https ://mybusiness.googleapis.com/v4/accounts/{accountno}/notifications

输出：

我在 pubsub 设置中也给予了必要的权限

使用 GCP 探索 ETL 过程。我在 Dataflow 中使用 Pub/Sub Subscription to BigQuery 模板。

Pub/Sub Subscription 中的消息数据为 csv 格式，如下所示

53466,06/30/2020,,特立尼达和多巴哥,2020-07-01 04:33:52,130.0,8.0,113.0

这会在加载到 BigQuery 表时出现错误。如何在模板中将 CSV 数据转换为 JSON？

我想使用 spring 2.5.5-SNAPSHOT，但是，我没有看到 GCP-Messaging 可用于该版本。这是否意味着它将在以后添加或弃用？

我的目标

1. 使用 GCP pubsub 或 pubsublite 设置 springboot 应用程序。

2. 包括“activemq”的启动器[将消息从兔子复制到 gcp-pubsublite]

3. 决定 gcp-starters 2.5.5 还是 2.4.5

新评论和谷歌更新通知即将到来时，谷歌问答的发布/订阅通知不会到来。请让我知道任何解决方案

我有一个 Dataflow 管道，它从 PubSub Lite 读取消息并将数据流式传输到 BigQuery 表中。该表按天分区。查询表时：

BigQuery UI 告诉我This query will process 1.9 GB when run。但是当实际运行查询时，我没有得到任何结果。我的管道现在运行了一整周，过去两天我得到了相同的结果。然而，2021-10-11在那之前的几天里，我看到了实际的结果。

我目前正在使用 Apache Beam 2.26 版，我的数据流编写器如下所示：

为什么 BigQuery 将值提交到分区需要这么长时间，但同时告诉我实际上有可用数据？

编辑1：

如何确认当前积压在 Google PubSub Lite 订阅中的所有消息。我尝试使用

但我不知道如何相应地设置偏移量。

我正在使用 Kafka 延迟主题消费consumer.pause(<partitions>)。

Pub/Sub Kafka shim 将 pause 变成 NoOp：

https://github.com/googleapis/java-pubsublite-kafka/blob/v0.6.7/src/main/java/com/google/cloud/pubsublite/kafka/PubsubLiteConsumer.java#L590-L600

是否有任何关于如何将 pub sub lite 主题的消费延迟设定的持续时间的文档？

即我想使用来自 Pub/Sub Lite 主题的所有消息，但有 4 分钟的综合延迟。

这是我使用 Kafka 原生的算法：

• 称呼consumer.poll()
• 恢复所有分配的分区consumer.resume(consumer.assignment())
• 将以前delayed的记录与最近轮询的记录结合起来
• 将记录分开
  • 足以处理的记录
  • 记录还太年轻，无法处理
• 为任何太年轻的记录暂停分区consumer.pause(<partitions of too young>)
• 保留一个太年轻的记录缓冲区，以便在下一次通过时重新考虑，称为delayed
• 处理足够老的记录
• 冲洗，重复

我们只提交足够老的记录的偏移量，如果进程死亡，“太年轻”缓冲区中的任何记录都将保持未提交，并且在随后的重新平衡中接收分区的任何消费者都会重新访问它们。

这种算法是否有更通用的形式可以与原生 Kafka 和 Pub/Sub Lite 一起使用？

编辑：CloudTasks 在这里是个坏主意，因为它断开了偏移提交链。我需要确保我只为从下游系统得到确认的记录提交偏移量。