《kafka》专题

Spring引导加载属性失败。下面是我通过yaml文件使用的属性。 我得到的例外是这个 原因：java.lang.IllegalAccessException:Class org.apache.kafka.common.utils.utils无法访问带有“protected”修饰符的类org.springframework.kafka.support.serializer.JsonDeserial

我以前是学习Kafka的传统ActiveMQ用户。我有一个问题。 使用Active MQ，您可以执行以下操作： 将100条消息提交到队列中 我试着在Kafka做同样的事情 如果不启动Consumer，等待它启动，然后运行producer，则此示例不起作用。 谁能告诉我如何修改我的示例程序，以在消息等待被消费的地方执行操作？

当使用循环调用kakfa生产类时，无法将消息写入kafka主题（生产者）。 我对Python和Kafka很陌生。我正在尝试编写一个python程序，将消息写入Kafka主题并生成，以便Kafka消费者可以订阅该主题以发布消息。 我不确定我的程序中缺少了什么，它限制了我将信息写入主题。 请注意：我正在读取一个JSON文件，并使用for循环来准备键值。然后将其分配给一个变量，并使用Kafka prod

我试图为每个绑定设置valueSerde，但是只考虑默认的valueSerde。 应用程序类 StreamsConfig。JAVA 和应用。yml 非常感谢您对这方面的任何见解。 编辑 完整的示例代码在这里 堆栈跟踪020-12-06 21:55:39.929错误141897---[-StreamThread-1]

我正在使用Spring Kafka consumer和Avro模式构建我的应用程序。 但是，如果消息无法反序列化到我构建的指定Avro特定记录，消费者将不断地反复尝试相同的消息（无限重试）。 在这种情况下，如果我的使用者出现反序列化程序异常，我如何配置使用者应用程序以跳过当前消息并移动到下一个偏移量。 我已经研究了Spring Kafka错误句柄，它只能处理侦听器中的异常，而不是在反序列化阶段。

我第一次使用Spring Kafka，我无法在我的消费者代码中使用Acknowledgement.acknowledge（）方法进行手动提交。请让我知道我的消费者配置或侦听器代码中是否缺少任何内容。或者有其他方法可以根据条件处理确认偏移。在这里，我正在寻找解决方案，例如如果偏移没有手动提交/确认，它应该由消费者选择相同的消息/偏移量。 配置 听众

我们有一个Akka应用程序，它使用Kafka主题，并将收到的消息发送给Akka参与者。我不确定我编程的方式是否充分利用了Akka Streams内置背压机制的所有优点。 以下是我的配置。。。 这做了我所期望的商业案例，myActor收到命令更新（MyAvro） 我更讨厌背压的技术概念，据我所知，背压机制部分由水槽控制，但在这种水流配置中，我的水槽只是“水槽”。“忽略”。所以我的水槽可以缓解背压。

用例如下。我在Java代码中的许多对象实例上传递生产者或消费者引用。在其中一些地方，我想对Kafka的配置进行一些检查。这意味着我想回去，Kafka生产者/消费者（包括默认值）中存储了什么样的有效配置。我在java文档中没有看到显式的anthing： Kafka制作人 那么，如何找回Kafka制作人和消费者的配置呢？

我可以在命令行上针对Kafka位置安装发送和接收消息。我也可以通过Java代码发送消息。这些消息显示在Kafka命令提示符中。我还有一个Kafka消费者的Java代码。代码昨天收到了消息。但是今天早上没有收到任何消息。代码没有更改。我想知道属性配置是否不太正确。这是我的配置： 制片人： 生产记录设置为 消费者： 对于Java代码： 少了什么？

我是Kafka的新手。我在网上读了很多关于Kafka制作人和Kafka消费者的说明。我成功地实现了前者，它可以向Kafka集群发送消息。然而，我没有完成后一个。请帮我解决这个问题。我看到我的问题像StackOverflow上的一些帖子，但我想更清楚地描述一下。我在虚拟盒子的Ubuntu服务器上运行Kafka和Zookeeper。使用1个Kafka集群和1个Zookeeper集群的最简单配置（几乎是

我有一个Kafka主题（1.0.0），只有一个分区。消费者被封装在EAR中，当部署到Wildfly 10时，最后一条消息的轮询始终返回0条消息。虽然主题不是空的。 当我做民意测验时，我得到0条记录。尽管日志记录显示： 当我更改为-2时，如： 我确实收到一条消息： 当然，这不是正确的记录，消息377408在哪里？ 尝试了许多方法来寻求结束等，但它从来没有工作。 这是我的消费者配置： 注意：我尝试了r

我在Scala中设置了Spark Kafka Consumer，它接收来自多个主题的消息： 我需要为每个主题的消息（将采用JSON格式）开发相应的操作代码。 我提到了以下问题，但其中的答案对我没有帮助： 从spark中的Kafka消息获取主题 那么，在接收到的DStream上是否有任何方法可用于获取主题名称以及消息以确定应该采取什么行动？ 对此任何帮助都将不胜感激。谢谢你。

我想知道是否可以在Kafka制作程序中配置2个不同的Kafka集群。 目前我正试图让我的制片人 我正在使用Apache Kafka 2.8和Python 3.7的confluent_kafka==1.8.2包。 生产商代码下方： 当我杀死clusterB时，我得到了以下错误消息。

我试图使用apache kafka二进制文件中的kafka控制台生成器生成消息，并在spring boot中使用消费者设置。消费者使用avro模式。 当消息以json格式生成时，我的消费者抛出异常-“无法序列化”。 我找到了一个解决方案，可以使用“ConFluent Platform 7.1”，它有kafka-avro-console-生产者。它支持avro，但它是企业版。 有没有一种方法可以使用

有什么方法可以阻止Kafka的消费者在一段时间内消费信息吗？我希望消费者停止一段时间，然后开始消费最后一条未消费的消息。