《kafka》专题

我在Kafka中有这样的配置 我有一个类，有相同的有效负载代码，我如何在任何时候在对象中转换它？ 大家好，我有一个微服务接收这个字符串并且正在工作，但是我需要将这个字符串转换为一个特定的对象，当我使用ObjectMapper转换应用程序时，会返回这个异常： 抛出异常；嵌套异常是com.fasterxml.jackson.databind.exc.InvalidDefinitionException

类和是常规POJO。 环境是： Java 11 Spring Boot 2.4.1 Kafka2.6.0 **第二次尝试-消费者和生产者（只是将键序列化/反序列化器设置为Json）** 例外

问题出现在Kafka的配置需要序列化程序。不起作用，JsonSerializer也不起作用，而字符串序列化程序将所有转义为，这使得使用这些消息非常麻烦。 如何生成这些JSON对象？是否有一个简单的Kafka序列化程序配置？

我们的微服务使用JDK8、Spring Cloud Stream API和Spring Boot 1.5.2版本，用于在Kafka队列上生成消息。对于消费者API，我们使用Spring Kafka 1.1.2.release，特别是。消息类型为应用程序json 谢谢

我想将Jackson的ObjectMapper注入到Kafka的ConsumerFactory的JsonDeserializer中。我已经做了这篇文章中提到的： 将ObjectMapper注入Spring Kafka序列化/反序列化程序 如何修复此问题并正确配置Kafka？

Kafka使用者(org.apache.Kafka:kafka-clients:2.3.0)应该使用来自一个分区的主题的消息。 该服务记录使用者订阅，但它不使用消息，也不使用官方的Apache工具显示在使用者组列表中。

具有特定组id的使用者连接到代理，监听主题不到1分钟，然后断开连接（根据业务逻辑）。当它监听主题时，它可以使用一些消息。当同一个使用者重复这个动作时，它会使用相同的消息！ 我发现Kafka用间隔1分钟保存偏移。这意味着消费者必须听超过1分钟的主题。我怎样才能缩短这个间隔？ 我发现了这样的属性： null null

我正试图为以下内容做POC： 但是，它不能从我的本地应用程序连接到Azure上的Kafka。 你能建议我在这里做错了什么吗？我将如何从本地应用程序连接到运行在azure（或任何云）上的容器中的kafka？

我正在使用来使用来自spring-boot应用程序中某个主题的消息，我需要定期运行该应用程序。spring-kafka版本是2.2.4.发行版。

null

我想找出将消息从Kafka路由到连接到负载平衡应用服务器集群的web套接字客户端的最佳方法。我知道spring-kafka有助于消费和发布消息到kafka主题，但是当连接到分布式kafka主题时，这在负载平衡的应用服务器场景中如何工作。以下是我希望满足的需求，其总体目标是在具有非常非常大的用户量的应用程序中促进对等消息传递： null 我认为，应用服务器负载平衡可以通过将具有特定路由密钥的用户（用

我想使用spring-Kafka库使用spring boot配置的消费者来使用来自Kafka代理的消息，源是一个JDBC连接器，它负责从MySQL数据库提取消息，这些消息需要被使用 下面是我的application.yml文件

我有一个有8个分区的Kafka主题，从单个消费者订阅该主题，我有唯一的消费者组为消费者。现在，我尝试只使用来自所有分区的最近消息（在我的例子中，是从当前时间开始的3分钟前）。我使用了如下所示的offsetForTimes方法。 现在的问题是offsetsForTimes只返回一个或两个分区偏移量位置，并返回null表示剩余位置。 我想使用所有分区最近的消息，而不是一个或两个分区。 我在下面也试过了