《kafka》专题

澄清：请注意，这个问题与这个问题不同：如何使用Spring Cloud Stream Kafka和每个服务的数据库实现微服务事件驱动架构 这一个是关于使用Kafka作为唯一的存储库（事件），不需要DB，另一个是关于使用每个服务的数据库（MariaDB）。 我想实现一个事件源架构来处理分布式事务： OrdersService接收订单请求并将新订单存储在代理中。 这是我的主要疑问：我如何查询Kafka

我有一个用Spring Boot构建的Camel应用程序。此应用程序使用ActiveMQ在两个系统之间传递消息。现在，我需要根据配置添加在ActiveMQ和Apache Kakfa之间选择作为消息代理的能力。 这是可以做到的吗？我对Kafka一无所知，但就我所阅读的内容而言，它的工作方式与ActiveMQ不同。 假设我的Camel应用程序向这样的队列发布消息： 我的申请表中有。yml此配置 我能用

我想知道Kafka选项重要性级别的区别。阿帕奇·Kafka有三个等级。 组织。阿帕奇。Kafka。普通。配置。配置定义。重要性 这三者有什么区别？

我们将收到处理该物品属于以下任何组的请求- Grp1- 我们收到3个A1、B1、C1请求。由于所有请求都属于同一组，因此应按顺序处理这些请求（在提取下一个请求进行处理之前，应先完成当前请求） 我们收到了4个A1、A2、B2、B3请求，其中A2和B2请求属于同一组，因此应按场景1中所述顺序处理，A1和B3请求应并行处理，因为它们属于不同的组 因此，基本上应该并行处理来自不同组的请求。来自同一组的请求

我发现maven repo中有几个Kafka。 阿帕奇的maven回购协议中有两个Kafka。https://mvnrepository.com/artifact/org.apache.kafka/kafka https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients 它们都可以从kafka服务器生成Mesg并消耗msg。 我

我目前正在尝试编写一个适配器，它将使用来自ActiveMQ的消息并将其发布到Kafka。 我正在考虑使用Spring集成来集成这两个消息传递系统。 我的问题是，我的应用程序不会维护模型的注册表，许多应用程序将使用该注册表将记录发布到activeMQ。我想接收这些javax-jms消息，并想执行一些转换，比如将jmscorrelationId添加到kafka消息中。 另外，另一个要求是仅当kafka

我目前正在开发Kafka模块，我正在使用Kafka通信的抽象。我能够集成生产者 Spring Boot测试类 监听器类 我的问题是：在测试类中，我断言分区、有效负载等是从BlockingQueue轮询的，然而，我的问题是如何验证用KafkaListener注释的类中的业务逻辑是否得到正确执行，并根据错误处理和其他业务场景将消息路由到不同的主题。在一些示例中，我看到了CountDownLatch的断

我决定回到我的想法，将Kafka指标与Spring Boot Actuator集成，我在这里已经提到： https://github.com/spring-projects/spring-boot/issues/6227 到目前为止，我有一个单独的“沙盒”项目和工作代码，我想将其合并到Spring Boot中。现在我有点困惑了。我的部分测试需要powermock来模拟Kafka的“超级安全”类：

我正在使用带有kafka绑定的spring cloud streams。消费者使用spring进行配置。云流动Kafka。绑定。功能配置和DLQ已启用。 当我使用来自主题的消息并发生异常时，我看到它重试3次，并将消息发送到DLQ。我已将maxAttemts配置为5，但我无法覆盖默认值3。 我使用的是spring kafka（2.7.8）和spring cloud（2020.0.4）。如何覆盖重试尝

我正在使用Kafka活页夹的Spring Cloud Stream。它工作得很好，但客户端接收到重复的消息。已经尝试了所有Kafka消费属性，但没有结果。 在我的应用程序示例中检查2个类-Aggregate Application和EventFilterApplication。如果我运行EventFilterApplication-只有1条消息，如果是Aggregate Application-2

我正在做一个Spring Boot应用程序，并试图以编程方式配置kafka，但由于某些原因，我仍然在从应用程序获取属性。yaml而不是我通过编程设置的

我有一个使用Spring Cloud Stream（v1.3.0）和Kafka（v1.1.6）的Spring boot（v.1.57）应用程序。我希望能够优雅地关闭它，即关闭时，所有流监听器（即用@StreamListener注释）应该： 停止轮询新消息 完成他们的工作 将偏移提交给Kafka 我注意到ContainerProperties（默认设置为10000ms）中有一个名为“Shutdown

需要帮助才能获得帮助，将GCP（Google Cloud Pubsub）的所有消息发布到远程运行的Kafka主题 我使用了“https://powerspace.tech/how-to-stream-data-from-google-pubsub-to-kafka-with-kafka-connect-dbef1c340a76”中的连接器代码，并且能够将数据从GCP移动到Kafka主题。Kafka

我试图运行一个Spring boot Kafka流示例https://cloud.spring.io/spring-cloud-stream-binder-kafka/spring-cloud-stream-binder-kafka.html#_usage_2地点 我能够成功地构建它。但在运行时，出现如下所示的错误（java.lang.IllegalStateException:org.sprin

如何更改@kafkastreamssstatestore的默认serdes？我知道，在Kafka流云的新版本3.0.1中，方法如下所述：https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.1.RELEASE/reference/html/spring-cloud-stream-binder