apache-kafka 相关问题

如何在已运行的集群上增加 __consumer_offsets 复制数量（Kraft kafka 3.4.0）

我想分享我对这个主题的发现，我希望这可以帮助遇到这个问题的人。 现在我们都应该知道如何增加主题复制因子，如果你不知道......

kafka 经纪商领导者倾斜

我有10个节点的kafka集群。kafka代理和zookeeper都在每个节点上运行。最近我们添加了 3 个新节点 8、9 和 10，昨天有 2 个节点关闭（2 和 4）。我有一个 60 的主题

KafkIO：无法在架构注册表中创建新的“旧”版本架构，没有错误

我正在 KafkIO 中创建新模式，有时无法创建新模式。最简单的例子，假设我创建了一个简单的模式，版本 1： 创建简单的架构 然后我创建一个新方案...

如何查看Kafka消息？

有什么方法可以查看给定主题发送到 Kafka 的消息内容吗？如果可能的话，请说一些类似查看该主题的最后五条消息之类的内容。

spring Kafka 在单个消费者中获取随机偏移量和分区？

我们有一个消费者。 我们正在记录每条收到的消息的偏移量和分区。 @KafkaListener(topics = "${kafka.topic.mytopic}") 公共无效消耗（@Payload字符串消息...

Confluence 连接器任务失败

启动docker容器（cp-kafka-connect-base：7.0.1）并安装自我管理连接器（debezium-connector-mysql：最新）。 Docker 容器运行良好。 当我尝试配置

将 repartition.purge.interval.ms 配置设置得太高以防止清除是否有任何副作用？

我们在应用程序中使用 Kafka Streams 来处理事件。 为了加入操作，我们使用 Kafka Streams 的 repartition 和 selectKey 方法，这会导致创建内部修复...

即使 auto.commit.offset 为 false，在 Kafka Consumer 中设置 group.id 是否会导致自动偏移同步？

每当在消费者属性中配置 groupId 时，无论 auto.commit.offset 设置为 true 或 false，我都会在轮询时获得偏移量消耗之间的同步。当投票时...

Postgres 的 Debezium 连接器不适用于分区表

我使用 Debezium 连接器设置了 Postgres->kafka CDC。 Debezium CDC 能够针对常规 postgres 表发布对 Kafka 主题 public.some_topic 的所有更改。 但它并没有发布对...的更改

Spring Boot：Kafka 健康指标

我有类似下面的东西，效果很好，但我更喜欢在不发送任何消息的情况下检查运行状况（不仅仅是检查套接字连接）。 我知道 Kafka 有类似 KafkaHealthIndi 的东西...

在实施“事务发件箱模式”时管理快速增长的“发件箱”表

我的服务需要发布消息作为更新数据库的事务的一部分。我正在使用 Postgres 和 Kafka，并且我决定采用事务发件箱模式：https://

Kafka Stream KTable 外键左连接输出比预期更多的墓碑

迁移到最新的流版本 3.9.0 后，我注意到左外键连接中有一个我无法理解的行为。 对于左外键连接： KTable ...

如何在php kafka中只获取最新记录

我使用 arnaud-lb/php-rdkafka 作为 PHP kafka 客户端。我使用的是 Windows 10 和 php 7.4。我试图通过消费者仅获取最新记录，但它正在返回所有记录。我不确定...

Spring Kafka Consumer 在 LeaveGroup 请求后无法重新加入

我们在高负载的生产环境中使用Spring Kafka。 我们使用了 @KafkaListener 注释并创建了这些侦听器作为 Spring Boot 服务的一部分。 这些消费者经常发送

如何从Kafka中的特定生产者中提取消息

我有两个Kafka队列（queue1和queue2），并且多个生产者（客户端）正在将消息发布到两个队列。每个客户端向队列发送不同类型的消息（例如，client1 send &q...

如何将 Oracle CDC Source Connector 生成的 CDC 事件提取到 Snowflake 中

我们当前的管道遵循与此处概述类似的结构，只是我们从 Oracle 中提取事件并将其推送到雪花。流程是这样的： 汇合...

通过java应用程序发布消息时与Kafka代理的连接错误

我正在尝试在 Ubuntu（在我的 Windows 笔记本电脑上）上运行的 Kafka 代理上发布运行 Java 应用程序（在我的 Windows 笔记本电脑上）的 kafka 消息。 我得到的错误是： [kafka-生产者-网络-...

Kafka Connect REST 接口“PUT /connectors/(string: name)/config”返回错误代码 500

我有一个分布式模式的 3 节点 kafka-connect 工作集群，带有正在运行的 s3 接收器连接器。要在运行时更新连接器的配置，我运行以下命令： 卷曲-X P...

如何维护Kafka中消息的顺序？

我们有消息层的要求，我们希望保留消息的顺序，直到它被消费者消费为止。我们正在探索卡夫卡的同样目的。我知道卡夫卡不保证顺序...

'org.apache.flink.formats.avro.AvroDeserializationSchema org.apache.flink.formats.avro.AvroDeserializationSchema.forGeneric(org.apache.avro.Schema)'

嘿，我在这里遇到了同样的问题，使用 avro 和模式注册表让 flink 从 kafka 读取数据。我可以使用 flink 尝试从服务器读取数据来查看 schema-registry 上的日志，但是......