问题标签 [kafka-topic]

我需要创建一个 Kafka 设置，在每天开始时，必须删除主题中的所有消息并将其偏移量重置为 0。

我试图通过将 log.retention 设置为 1 天来做到这一点，但它似乎没有用。我也跑了kafka-delete-records。它确实删除了所有消息，但kafka.tools.GetOffsetShell仍然给了我最新的偏移量。我希望它变为 0。

另外，我没有明确设置消费者组。它们是内部的，例如_confluent-ksql-default_transient_929377247691956102_1560973030312. 因此，无论消费者组如何，我都希望将主题的偏移量重置为 0。

假设我有一个 kafka 主题，大约有 10 个分区，我知道每个消费者组在任何给定时间都应该有 10 个消费者从主题中读取，以实现最大的并行性。

但是，我想知道对于一个主题在任何给定时间点可以处理的消费者组的数量是否有任何直接规则。（我最近在一次采访中被问到这个问题）。据我所知，这取决于代理的配置，以便在任何给定时间点它可以处理多少个连接。

但是，只是想知道在给定时间点可以扩展多少个最大消费者组（每个有 10 个消费者）？

我在 springboot 应用程序中有一个简单的 kafka 消费者监听器，如下所示：

在某些特定情况下，我想拒绝该消息，但我希望系统在一定时间后再次向我建议；

我能怎么做？

注意：我也想定制kafka配置（不是默认的springboot结构）

通过 Kafka 文档和其他各种资源，我了解到 Kafka 中的消息被组织成主题。此外，一个主题可以分解为多个分区，每个分区可以托管在不同的服务器上。这提供了冗余和可扩展性。

我不确定“破碎”这个词在这里是什么意思。这是否意味着如果添加到主题的消息是“1 2 3 4 5 6 7”，那么在将其分成多个分区后，我们将有一个分区只有整个主题的一个子部分。就像一个分区有'1 2 3'，而另一个分区有'4 5 6'，而另一个分区只有'7'。或者这是否意味着每个分区都有“1 2 3 4 5 6 7”，这意味着我们有精确的副本。

我尝试在 Windows 中使用时间戳模式创建源 Kafka 连接。我有这张桌子：

oracle数据库属性：

控制台消费者不返回任何输出：

连接器仅在批量模式下工作。我不懂为什么。

检查连接器：

Ed：在调试模式下，我的输出是这样的。

源卡夫卡连接：

输出：

如何控制时差？

我了解我的默认保留期，但我想查看是否有任何主题对保留期有特殊覆盖。

我试过使用 kafka-topics，但它没有与 zookeeper 连接。

kafka-topics --zookeeper localhost:2181 --describe --topics-with-overrides

我希望这个输出基本上没有，但我收到错误消息：

像往常一样，看到拆分方法优于其他方法的好处有点令人困惑。

1. 我看不出拥有之间的区别/利弊
   • Topic1 -> P0和Topic 2 -> P0
   • 一个消费者从 2个Topic 1 -> P0, P1
     主题或单个主题/2 个分区中拉取，同时P0将P1持有不同的事件类型或实体。

你唯一的好处我可以看到如果另一个消费者需要主题 2 数据，那么它很容易消费

2. 关于主题自动生成，这种方式背后有什么好处，或者一段时间后它会失控吗？

谢谢

I have this use case that I have 2 topics

Topic 1 (Units) -> P0 / Topic 2 (Reservations) -> P0

I have a single consumer that needs to have the up to data from both topics/partitions in order to take the correct decision (either delete a unit if not reserved or reserve the unit if it exists)

I decided to have them in 2 different topics for concept segregation as well as at any single time I would need a 3rd service to interact with units only then that's possible in an easy way.

But the idea of how to handle the concurrent operation that could arise or maybe delayed events from a topic over another topic?

Thanks

我有一个用例，我必须将所有 MySQL 数据库数据推送到 Kafka 主题。现在，我知道我可以使用 Kafka 连接器启动并运行它，但我想了解它是如何在不使用连接器的情况下在内部工作的。在我的 Spring Boot 项目中，我已经创建了一个 Kafka Producer 文件，我在其中设置了所有配置，创建了一个 Producer 记录等等。

有没有人尝试过这种方法？任何人都可以对此有所了解吗？

我是 Kafka 的新手，并试图在我的本地机器上创建一个新主题。我正在关注这个https://medium.com/@maftabali2k13/setting-up-a-kafka-cluster-on-ec2-1b37144cb4e

1. 启动动物园管理员

启动 kafka-server

创建主题

但是在创建主题时，我收到以下错误

我看过以下为什么kafka没有创建主题？bootstrap-server 不是一个公认的选项 但是我在这里找不到我的问题的答案，因为给出的错误是不同的。我在这里缺少一些东西吗？