如何在启动上下文之前在内存中加载压缩的主题
我使用的是Kafka中的压缩主题，在应用程序启动时将其加载到HashMap中。 然后，我监听消息的普通主题，并使用从压缩的主题构造的HashMap来处理它们。 在开始收听其他主题之前，如何确保压缩的主题已完全阅读且HashMap已完全初始化？ (RestController相同) 推荐答案 实现SmartLifecycle，将地图加载到start()中。请确保phase早于任何其他需
..
spring-kafka相关内容
如何获得KafkaListener消费者
我使用SpringBoot 2.0.4.RELEASE的Spring-Kafka。 并使用KafkaListener获取消息 现在我要重置我的组的偏移量 但我不知道如何获得群的消费者 @KafkaListener(id="test",topics={"test"},groupId="group",containerFactory="batchContainerFac
..
创建不带注释的KafkaListener&；不带弹簧靴子
我正在尝试为一个主题创建一个Kafka Consumer，而不使用@KafkaListener注释。我之所以要这样做，是因为我试图在不使用SpringBoot的情况下，基于Application.Properties动态创建侦听器。 我想最好的方法是手动创建一个KafkaListenerContainerFactory。谁能提供一个在它自己的类中如何做到这一点的例子吗？ 推荐答案
..
Spring Kafka模型不在受信任的包中
我正在使用spring-Kafka-2.1.5和spring-boot-2.0.5处理微服务 第一个服务将向Kafka生成一些消息，第二个服务将消费这些消息，而消费时我遇到了问题 Caused by: java.lang.IllegalArgumentException: The class 'com.service1.model.TopicMessage' is not in the
..
多个消费者使用春卡夫卡
我希望在我的应用程序中设置多个Kafka主题侦听器。下面是我的设置。它应该由两个组使用，但只由一个监听器使用。我在这里错过了什么？ @Bean public Map consumerConfigs() { Map props = new HashMap(); props.put(C
..
如何将@Transaction与@KafkaListener一起使用？
有没有可能将声明性TX管理(通过@Transaction)与@KafkaListener注释方法一起使用？ 例如，我想使用它来为每个监听器定义单独的发送超时。 我的设置如下： TransactionManager： @Bean @ConditionalOnBean(value = {HibernateTransactionManager.class}) public ChainedKa
..
春天·卡夫卡听着regex
我正在尝试使用以下代码收听新创建的主题，但不起作用。您能告诉我下面的代码是否正确吗？ public class KafkaMessageListener { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageListener.class); private final Proce
..
在Spring Boot中覆盖Kafka版本因某些依赖而失败
我有一个新的Spring Boot 2.6.3 Java 11应用程序，它具有Spring Kafka依赖项(使用start.spring.io生成)。 默认使用Kafka 3.0.0。我想将卡夫卡版本改为3.1.0并添加 3.1.0 到属性部分。大多数Kafka库现在在版本3.1.0中都已解决，但不是所有(最重要的是，
..
卡夫卡客户(消费者/生产商)在崩溃后恢复
在我工作的公司，我们在没有身份验证的情况下使用Spring for Kafka，最近我们做了一些实验来设置Kafka中的安全性，并且我们在短时间内启用了身份验证，这导致了我们微服务中所有消费者/生产者的崩溃！(微服务保持正常) 异常： Authorization Exception and no authorizationExceptionRetryInterval set org.
..
我可以在运行时向我的@kafkalistener添加主题吗
我已经为主题数组创建了一个Bean，并在运行时向该主题数组添加了一些主题，但使用者没有更新主题，仍然从主题数组中的第一个主题开始消费。我希望消费者添加这些新主题并开始使用它们 @Autowired private String[] topicArray; @KafkaListener(topics = "#{topicArray}", groupId = "MyGroup") pu
..
Spring Cloud Kafka Streams中的错误处理
我使用的是Spring Cloud Stream和Kafka Stream。假设我有一个处理器，它是一个将字符串的KStream转换为CityProgrammes的KStream的函数。它调用一个API来按名称查找City，并调用另一个转换来查找该城市附近的任何事件。 现在的问题是转换过程中发生任何错误，整个应用程序都会停止。我想把这一条特别的信息发送给DLQ，然后继续前进。我已经读了几天了
..
如何配置Spring Boot Kafka客户端以使其不尝试连接
这与Is there a "Circuit Breaker" for Spring Boot Kafka client?有关，但我仍然认为这是一个不同的问题：) 我们需要配置Spring Boot Kafka客户端，以便它根本不会尝试连接。 用例是，在测试环境中，我们没有运行Kafka，但我们仍然需要构建完整的Spring Boot上下文，因此使该Bean以配置文件为条件是不起作用的。
..
是否有适用于Spring Boot Kafka客户端的断路器？
如果Kafka服务器(暂时)关闭，我的Spring Boot应用程序ReactiveKafkaConsumerTemplate一直尝试连接不成功，从而造成不必要的流量和日志文件混乱： 2021-11-10 14:45:30.265 WARN 24984 --- [onsumer-group-1] org.apache.kafka.clients.NetworkClient : [Con
..
如何在Spring-Kafka中实现Consumer SeekAware？
我正在尝试用@KafkaListener实现一个消费者。 我使用的是Spring2.3.7版本。 以下是我到目前为止的代码 public class SampleListener { @KafkaListener(topics = "test-topic", containerFactory = "sampleKafkaListenerContainerFactory
..
SSE`this.ventSource.onMessage`调用失败。错误`事件源的响应具有不是文本/事件流的MIME类型(应用程序/json"；)。
ANGLE Server Sent事件this.eventSource.onmessage调用失败，错误"EventSource's response has a MIME type ("application/json") that is not "text/event-stream". Aborting the connection."我在Chrome Dev工具(附图)中看到有两个Conte
..
如何编写单元测试用例为 ListenableFuture 添加回调
我正在尝试为 ListenableFuture 添加 Callback 编写单元测试用例，但我不知道该怎么做.在互联网上没有得到任何有用的东西. @Test公共无效 can_publish_data_to_kafka() {字符串主题 = someString(10);字符串键 = someAlphanumericString(5);字符串数据 = someString(50);SendRes
..
如何编写单元测试用例来为 ListenableFuture 添加回调
我正在尝试为 ListenableFuture 添加回调编写单元测试用例，但我不知道该怎么做.在互联网上没有得到任何有用的东西. @Testpublic void can_publish_data_to_kafka() {字符串主题 = someString(10);字符串键 = someAlphanumericString(5);字符串数据 = someString(50);SendResu
..
如何在 spock 中测试 ListenableFuture 回调
几天前我问了一个关于从 kafka.send() 方法存根未来响应的问题.@kriegaex 此处回答并正确解释了这一点虽然我遇到了另一个问题，但我如何测试这个未来响应的 onSuccess 和 onFailure 回调.这是正在测试的代码. import org.springframework.kafka.core.KafkaTemplate;导入 org.springframework.ka
..
如何在java中更改第3方库的日志级别
控制台日志中堆满了来自第 3 部分库的日志.例如，我的项目使用 kafka 和 zooker keeper 客户端库，因为它们的日志太多 2018-05-08 10:30:38.250 INFO 2968 --- [0:0:0:0:1:2181)]org.apache.zookeeper.ClientCnxn:打开套接字连接到服务器 0:0:0:0:0:0:0:1/0:0:0:0:0:0:0
..
Kubernetes 中的 Kafka - 将协调器标记为组死亡
我对 Kubernetes 还很陌生，想用它来设置 Kafka 和 zookeeper.我能够使用 StatefulSets 在 Kubernetes 中设置 Apache Kafka 和 Zookeeper.我关注了 this 和 this 来构建我的清单文件.我分别制作了 1 个 kafka 和 zookeeper 的副本，并且还使用了持久卷.所有 Pod 都在运行并准备就绪. 我试图通
..