如何访问kafka主题中所有未提交消息的列表

我使用 Apache Kafka 作为消息处理器，并在 Asp.Net Core 中使用 Confluent.Kafka 作为消费者。我必须使用几个消费者来处理来自主题的所有消息。当我没有任何异常时，我的代码可以完美运行。2 个消费者并行工作。如果我在消息处理过程中出现异常，就会出现问题。有时（实际上并不经常）我有偏移量重复，这会导致重新处理已处理的消息。

这是代码，它模拟处理过程中抛出的随机异常。再有一次，如果我删除随机异常并启动 2 个消费者，一切正常。但除了我有偏移重复...

这是批处理的扩展：

我应该怎么办？

在主题上调用 Subscribe(topicname) 和调用 Assign(TopicPartition) 有什么区别？

两者都尝试后，我发现我仍然可以调用 Consume() 并且消费者会收到可用的消息。

我注意到的一个区别是，在订阅时，消费者在超过最大轮询间隔后得到（EAGER）重新平衡。使用 Assign(TopicPartition) 时不会发生这种情况。如果有人也可以详细说明这一点。

我正在尝试解决 Kafka 连接问题。基本上我正在尝试从 .net core confluent kafka 到 kafka 集群（linux服务器）的生产/消费，我面临着各种问题。我正在使用 Confluent.Kafka(.net) 1.7.0 版本连接到 Kafka 集群版本：2.7.2 我的 confluent kafka .net 正在 docker 容器中使用。

我正在尝试在 docker 容器中运行我的 .net core c# api，但出现以下运行时错误：

任何人都对这类问题有一些经验。

更新：我做了，我最终关注了 [link] Confluent Kafka Dotnet Kerberos Support Dockerfile（没有 SASL 机制 GSSAPI 的提供者）文章。
对 dockerfile 进行了更改以完成这些步骤。在容器上安装后，我收到以下消息：

当用户尝试使用 Kerberos 并指定主体或用户名而不指定主体所属的管理 Kerberos 领域时，系统会附加默认领域。默认领域也可以用作在本地机器上运行的 Kerberos 服务的领域。通常，默认领域是本地 DNS 域的大写版本。

我对 Kafka “本地：值序列化错误”有疑问。

问题

我的应用程序通过掩码从多个主题中读取消息，并因错误而崩溃：

我试过的

第一个问题是我遵循了处理大消息的准则：fetch.max.bytes，message.max.bytes并且receive.message.max.bytes配置设置为限制：

但是应用程序仍然无法处理该消息。所以我决定跳过这条消息

第二个问题是我找不到导致错误的消息。我已确定已停止处理的主题（CURRENT-OFFSET未更改LAG且不等于 0）。我检查了这些CURRENT-OFFSET消息 - 它们都很小：

问题

如何识别导致错误的消息？

我猜一个解决方案将使用命令行实用程序输出具有分区/偏移量的主题的最大消息大小

I'm trying to create a topic (an Event Hub) programmatically from the Kafka interface using AdminClient.CreateTopicsAsync. This works when connecting to Kafka, but not to Event Hub. I'm running into the following error:

Default partition count (KIP-464) not supported by broker, requires broker version <= 2.4.0

It complains that using a default number of partitions is not supported, but as far as I can tell, I can't provide one as the underlying librdkafka does not support it. The only information I could find by googling this is that someone in 2021 did make it work.