Spring Cloud Stream + Kafka动态目的地+错误处理
Spring Cloud Stream是一个用于构建消息驱动微服务的框架，它基于Spring Boot和Spring Integration，提供了一种简化和标准化的方式来处理消息。
Kafka是一个分布式流处理平台，它具有高吞吐量、可扩展性和持久性的特点。它通过将数据分区和复制到多个节点来实现可靠性和容错性。
动态目的地是指在运行时动态地创建和删除消息队列或主题。Spring Cloud Stream提供了动态目的地的支持，可以根据需要动态地创建和删除Kafka主题。
错误处理是指在消息处理过程中出现错误时的处理机制。Spring Cloud Stream提供了多种错误处理机制，包括重试、错误通道和死信队列等。
Spring Cloud Stream + Kafka动态目的地+错误处理的应用场景包括：
- 实时数据处理：通过使用Spring Cloud Stream和Kafka，可以构建实时数据处理系统，将数据从生产者发送到消费者，并在处理过程中进行错误处理和重试。
- 异步通信：使用Spring Cloud Stream和Kafka，可以实现异步消息通信，将消息发送到Kafka主题，并由消费者进行处理。
- 日志收集和分析：通过将日志消息发送到Kafka主题，可以实现日志的集中收集和分析，以便进行故障排查和性能监控。
腾讯云提供了一系列与消息队列和流处理相关的产品和服务，可以与Spring Cloud Stream和Kafka集成使用。以下是一些推荐的腾讯云产品和产品介绍链接地址：
- 云消息队列CMQ：https://cloud.tencent.com/product/cmq
- 云原生消息队列TDMQ：https://cloud.tencent.com/product/tdmq
- 云流计算Ckafka：https://cloud.tencent.com/product/ckafka
- 云函数SCF：https://cloud.tencent.com/product/scf
请注意，以上答案仅供参考，具体的产品选择和配置应根据实际需求和情况进行评估和决策。
相关·内容
2回答
在尝试使用一个@StreamListener中的多个主题时，我遇到了一个spring引导流的问题。
根据春云流文档:确认到文档：
目标
绑定中间件上通道的目标目的地(例如，RabbitMQ exchange或Kafka主题)。如果信道绑定为使用者，则可以将其绑定到多个目的地，并且可以将目标名称指定为逗号分隔的字符串值。如果未设置，则使用通道名称。不能重写此>属性的默认值。
但是，在我使用了next config之后：
spring:
cloud:
stream:
bindings:
testchannel:
group:
浏览 4提问于2017-06-18得票数 5
回答已采纳
我想创建一个kafka端点使用spring cloud stream框架，它将有一个http post api到/{topic_name}.How我可以动态改变spring.cloud.stream.bindings.<channelName>.destination={topic_name}属性吗？
我可以使用confluent Kafka REST Proxy实现来实现上述功能，但我想知道是否有可能在春季开发此功能。
浏览 20提问于2020-02-09得票数 0
回答已采纳
在使用@StreamListener收听kafka主题之后，在RuntimeException、全局erroChannel或主题特定errorChannel (topic.group.errors)上没有收到任何错误消息。@ServiceActivator没有收到任何信息。
POM依赖项: Greenwich.RELEASE
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactI
浏览 1提问于2019-06-07得票数 0
回答已采纳
我们有一个跳转应用程序，它从一个主题消耗，并生成记录到多个主题。
最近，在gradle项目中，这个应用程序升级到Sprinboot-2.6.7和其他依赖项。
应用程序能够正确消费和生产，，但问题是它似乎反复创建卡夫卡管理客户端(1000 S)，并且似乎正在泄漏内存(可能是由于这个原因？)，最终导致实例崩溃，无法跟上滞后。
外部库中与kafka相关的依赖jars
org.apache.kafka:kafka-clients:3.0.1
org.springframework.cloud:spring-cloud-stream:3.2.3
org.springfram
浏览 17提问于2022-08-18得票数 0
1回答
弹簧云函数路由
、、、、
我正试图通过实现Spring云功能来创建一个kafka。我使用的是spring版本2.4.0和spring版本：2020.0.0
@Slf4j
@Component
public class ArticleEventPublisher implements Function<ArticleAggregatedDocument, Message<ArticleMessage>> {
private final ArticleMessageMapper articleMessageMapper;
public ArticleEventPublisher(
浏览 4提问于2021-11-17得票数 0
回答已采纳
我是卡夫卡的新手，我想要帮助
我有两个应用程序(springboot)，它们是相同的/拷贝只有不同的端口。
他们都是消费者
这两个人听XXX这个话题
我有另一个应用程序扮演制片人的角色。
每当我给某某话题发东西。
只有其中一个使用消息，而另一个不使用消息。
我分别测试了两种方法，如果他们是单独听的话，他们也会正常地听，但是如果他们在一起听，只有一人会听。
我在用
<dependency>
<groupId>org.springframework.cloud</groupId>
<a
浏览 0提问于2021-07-06得票数 1
回答已采纳
当将事件发布到kafka主题时，我会遇到以下异常
org.apache.kafka.common.errors.RecordTooLargeException: The message is 2063239 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
我看了
我正在使用下面的spring云版本，没有在我的pom中为spring-cloud-stream-kafka-binder提供任
浏览 0提问于2017-03-22得票数 0
回答已采纳
1回答
当使用动态生成的目标时，我希望禁用消息中的嵌入标头，类似于来自的示例(例如输出主题= dyntopic1、dyntopic2、.)。
我已经设置了像下面这样的属性，但是我仍然得到标题，如果我遗漏了什么，有什么建议吗？
spring.cloud.stream.bindings.output.group=test-ogroup
spring.cloud.stream.bindings.output.binder=kafka
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings
浏览 4提问于2018-01-31得票数 1
回答已采纳
我正在使用spring -cloud-stream:1.3.0开发spring启动应用程序。messages，spring-cloud-stream-binder-kafka:1.3.0.RELEASE.我使用spring集成dsl来拆分文件中的行，并使用beanio将行转换为json，要求将成功的json消息写入kafka主题，并将错误消息写入不同的kafka主题。下面是application.yml中的配置。
spring:
cloud:
stream:
kafka:
binder:
autoAddPartitions: true
浏览 2提问于2017-10-19得票数 1
我正在使用Spring Cloud Stream，并希望以编程方式创建和绑定通道。我的用例是，在应用程序启动期间，我收到要订阅的Kafka主题的动态列表。那么如何为每个主题创建一个频道呢？
浏览 0提问于2017-01-16得票数 6
1回答
正在尝试将Spring配置为在使用批处理模式时将错误消息发送到死信队列。但结果是在dlq主题中什么都没有。 我使用Spring Boot 2.5.3和Spring Cloud 2020.0.3。这会自动将spring-cloud-stream-binder-kafka-parent版本解析为3.1.3。 下面是application.properties： spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true
spring.cloud.stream.bindings.input-in-0.content-type=t
浏览 56提问于2021-08-03得票数 1
回答已采纳
1回答
我正在写一个Spring Boot (2.1.4)应用程序，试图使用Spring Cloud Streams for Kafka。 我要做的是维护一个主题的传感器列表(“传感器”)。OTOH，我有另一个主题的传入数据(“数据”)。我试图实现的是，当我获得一个传感器的数据时，我希望将其添加到传感器列表中。 为此，我从传感器主题创建了一个KTable<String, Sensor>，将温度主题映射到纯传感器的数据(在本例中是它的名称)，并与一个ValueJoiner进行外部连接，如果传感器存在，则保留传感器，否则使用读数的传感器。然后，我将结果写回sensors主题。 KTable&
浏览 29提问于2019-05-03得票数 1
类似于：
我在微服务器架构中启动batch时做了一个poc，我正在使用Spring batch和Spring Cloud Stream Kafka。我正在寻找一种方法来动态创建消费者(处理器)应用程序的多个实例。我看到可以使用以下命令定义许多实例
spring.cloud.stream.instanceCount=n
spring.cloud.stream.instanceIndex=[0, ..., n-1]
但是，我还没有找到一种动态更改instanceIndex值的方法。是否可以使用Spring Cloud Stream kafka动态修改此值。
谢谢你的帮助。
浏览 6提问于2021-05-08得票数 5
我知道如何使用命令式编程方法定义生产者，但我找不到如何使用函数式编程方法定义生产者。 我读了Spring Cloud Stream Binder的文档，但只发现如何定义消费者，或者消费者和生产者(例如，从主题中获取信息，转换数据并发送到另一个主题)。 因此，我不知道是否可以继续使用像@Input、@Ouptut这样的注释来定义单个处理器，在这一点上我非常困惑，因为库表明这些注释已被弃用，但我找不到示例或文档来定义一个简单的生产者来向特定主题发送信息。 谢谢! 文档链接：https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3
浏览 30提问于2021-01-22得票数 2
1回答
需要一些帮助，在集成卡夫卡和春天的云流。应用程序非常简单，有两个部分(作为单独的java进程运行)。
使用者将请求放入RequestTopic并从ResponseTopic获得响应。
生产者-从RequestTopic获取请求，并将响应放回ResponseTopic中。
我为使用者创建了RequestSenderChannel和ResponseReceiverChannel接口，为生产者应用程序创建了RequestReceiverChannel和ResponseSenderChannel。它们共享相同的yaml文件。根据文档，spring.cloud.stream.bindin
浏览 1提问于2019-12-04得票数 0
回答已采纳
我正在尝试在Spring Cloud数据流中配置DLQ。下面是Stream的定义以及我如何部署它
stream create --definition ":someTestTopic > custom-transform
--spring.cloud.stream.bindings.input.consumer.headerMode=raw | log --spring.cloud.stream.bindings.input.consumer.headerMode=raw" --name ticktran
stream deploy ticktr
浏览 0提问于2016-09-16得票数 0
我有一个常规的java Function；我试图绑定它：
它对web端点的输入
它的输出到一个卡夫卡主题。
当我在web上下文中使用我的函数时，它总是将Function的结果值单独返回给web客户机。我能做这样的事吗
spring.cloud.stream.bindings.input.binder=web
spring.cloud.stream.bindings.output.binder=kafka
我目前甚至试图将Function分解为2：
它的输入绑定到web客户端，其输出动态绑定到第二个函数(使用spring.cloud.stream.sendto.destina
浏览 1提问于2019-11-29得票数 4
回答已采纳
我一直在尝试为Spring cloud stream kafka实现重试逻辑，以便在向主题sample-topic生成事件时抛出异常，它会再重试两次。
我在application.properties文件中添加了以下配置
spring.cloud.stream.bindings.processSampleEvent.destination=sample-topic
spring.cloud.stream.bindings.processSampleEvent.content-type=application/json
spring.cloud.stream.bindings.processSa
浏览 25提问于2021-07-06得票数 1
我使用的是spring cloud stream kafka。我已经为kafka设置了如下属性：
spring.cloud.stream.bindings.output.contentType=application/x-java-object;type=com.example.demo.KafkaApplication.Message
eureka.client.service-url.default-zone=http://172.16.165.26:8761/eureka/
spring.cloud.stream.kafka.binder.zkNodes=10.6.22.20
#spri
浏览 0提问于2017-06-21得票数 0
1回答
我需要将使用的无效消息发送到DLT中，但在单独的引导服务器中。我目前有这样的配置： spring.cloud.stream.binders.some-kafka-binder.type=kafka
spring.cloud.stream.binders.some-kafka-binder.environment.spring.cloud.stream.kafka.binder.brokers=localhost:29092
spring.cloud.stream.bindings.processor-in-0.binder=some-kafka-binder
spring.cloud.st
浏览 22提问于2021-10-18得票数 1
1回答
目标
我必须为kafka流使用者设置一个组id，它与严格的命名约定相匹配。
在深入了解文档之后，我找不到一种有效的方法。由于我仍然认为我可能错误地理解了一些东西，所以我更喜欢在春季云流github存储库中打开bug问题之前，在这里打开一个问题供同行评审。
注：
在一年前就已经被问到了，但是这个问题还没有得到很好的回答，我希望我能对这个问题有更多的了解。
正式文档说明了什么(同时也基于警告消息)
从官方文档的几个来源，我看到在我的应用程序的application.yaml中，这应该是非常容易配置的。
文件指出，我可以：
使用spring.cloud.stream.kafka.default.
浏览 19提问于2022-08-10得票数 1
回答已采纳
我使用的是spring cloud stream。我有两个频道。一种是使用kafka cluster1，一种是使用cluster2。config就像。 spring.cloud.stream.default-binder=kafka
spring.cloud.stream.binders.kafka.type=kafka
spring.cloud.stream.binders.kafka.environment.spring.cloud.stream.kafka.binder.brokers=xxxx
spring.cloud.stream.kafka.binder.auto-add-pa
浏览 241提问于2019-04-17得票数 1
我使用的是spring cloud stream。我现在介绍了如何使用application.yaml和Kafka创建静态绑定。我想在java代码中创建一个新的绑定，或者修改application.yaml中定义的绑定的主题。例如 cloud:
stream:
default-binder: kafka
function:
definition: listenEnrichment
bindings:
listenEnrichment-in-0:
destination: input 我想在运行时将目标输入更改为其他主题。这是可能的吗？谢谢。
浏览 9提问于2021-02-04得票数 0
1回答
我有一个代码，它使用KafkaTemplate发送消息到不同的卡夫卡主题。Topic名称是从API接收的，不能提前获知，代码如下： var message = MessageBuilder.fromMessage(message)
.setHeader(KafkaHeaders.TOPIC, targetTopic)
.build();
kafkaTemplate.send(message); 现在我想重构现有的方法来使用Spring Cloud Stream，但它需要在application.propertie
浏览 23提问于2021-02-10得票数 0
我尝试使用spring cloud stream发送带有键、值对的消息。我找不到任何用于此的API。org.springframework.messaging.MessageChannel只有有效载荷作为发送function.Using Kafka模板的一部分，这是可以实现的。对于apache kafka来说，生成键、值类型的message.As KafkaTemplate的唯一方法是spring的一部分，我希望在spring cloud stream中有同样的抽象可用。请提个建议。
谢谢,
浏览 1提问于2020-02-18得票数 2
我用春云条纹Kstream。我测试一个主题&一个@StreamListner。没事的。
我为两个KStream输入修改我的代码。(两个@StreamListener)但是，弹簧云错误..。
***************************
APPLICATION FAILED TO START
***************************
Description:
The bean 'stream-builder-process', defined in null, could not be registered. A bean with that
浏览 0提问于2019-03-13得票数 4
回答已采纳
这两个活页夹有什么区别？他们的行为或多或少都是一样的--比如生产、加工和消费。 https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.RELEASE/reference/html/spring-cloud-stream-binder-kafka.html#_reference_guide <dependency>
<groupId>org.springframework.cloud</groupId>
&l
浏览 20提问于2020-07-19得票数 0
回答已采纳
我正在尝试弄清楚如何使用Spring Boot Streams和使用的MQueue。我可以连接到MQueue，但需要一个Could not provision topic 'queue///EMB_DEV_QUEUE'和MQJE001: Completion Code '2', Reason '2035'。我确实向管理员确认了这是一个队列，而不是一个主题。
我可以使用基于的MQQueueConnectionFactory使用一些示例代码进行连接，因此我知道MQueue正在工作。
这是我的程序。我已经成功地对Kafka使用了相同的模式。
@Enab
浏览 0提问于2017-04-22得票数 0
2回答
我有一个用例，需要在运行时生成多个Kafka主题/目的地。我试图组合，方法是使用来自Function类型的函数式bean的返回Flux<Message<T>>，并为每个Message设置标题spring.cloud.stream.sendto.destination，如描述的那样。我想出了以下实现：
@Bean
public Function<Person, Flux<Message<Person>>> route() {
return person -> Flux.fromIterable(Stream.of(pers
浏览 0提问于2020-06-07得票数 1
回答已采纳
我无法使用文档()中指定的语法更改通道(或绑定)的Serde。
假设我的通道是pcin，我理解我应该使用折叠属性spring.cloud.stream.kafka.streams.bindings.pcin.producer.valueSerde和spring.cloud.stream.kafka.streams.bindings.pcin.producer.keySerde来指示valueSerde和keySerde。
不过，我收到一个例外情况：
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer
浏览 1提问于2018-04-18得票数 0
回答已采纳
1回答
我有一个spring项目，它依赖于2.1.2版本的spring-cloud-starter-stream-kafka和spring-cloud-stream。它作为一个制片人。当它开始的时候，它一直在警告我：
2019-05-06 15:00:25.136警告28116可能找不到经纪人。
并抛出一个错误：
2019-05-06 15:01:10.280 ERROR 28116 --- [ask-scheduler-1] o.s.cloud.stream.binding.BindingService : Failed to create producer binding; retryi
浏览 0提问于2019-05-06得票数 0
回答已采纳
当试图启动一个具有Kafka绑定器活动的项目时，控制台中会提示以下内容：
org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle';
Caused by: java.lang.NoSuchMethodError: java.util.List.of(Ljava/lang/Object;Ljava/lang/Object;Ljava/lang/Object;Ljava/lang/Object;Ljava/lang/Object;Ljav
浏览 8提问于2022-05-26得票数 0
回答已采纳
1回答
我遇到了使用KStream组件的Spring云流应用程序的问题。它是侦听一个输入，并在处理完消息后将消息定向到一个输出。
它期待一个JSON字符串进来，并试图在到达时将其转换为Spring。在发送消息时会发生相反的情况。
问题是，当系统管理员想要用kafka-console-producer.sh测试主题时.并打印一个字符串
"lol“
在它中，整个Spring云流应用程序将在那里死亡，但有以下例外：
java.lang.RuntimeException: com.fasterxml.jackson.core.JsonParseException: Unrecognized t
浏览 0提问于2018-02-13得票数 1
回答已采纳
以下是我迄今为止在Server上成功完成的工作
我已经成功地在我的本地部署了SCDF服务器，并且我也使用了Kafka和Zoo门将配置参数，即
spring-cloud-dataflow-server-local-1.3.0.RELEASE.jar --spring.cloud.dataflow.applicationProperties.stream.spring.cloud.stream.kafka.binder.brokers=localhost:9092 --spring.cloud.dataflow.applicationProperties.stream.spring.c
浏览 2提问于2018-02-12得票数 0
回答已采纳
1回答
我有多个生产者将消息推送到我的kafka代理中。我想保护我的主题，这样生产者需要密码才能在主题中推送消息或任何形式的保护。
在spring boot中，可以看到以下属性。
spring.cloud.stream.kafka.binder.configuration.[security.protocol]=SSL
#spring.cloud.stream.kafka.binder.configuration.[ssl.truststore.location]=D:\\streaming\\MessagingQueingTier\\SSL\\kafka.client.truststore.jks
浏览 0提问于2020-08-22得票数 2
我的项目application.properties发出警告。说
'spring.cloud.stream.kafka.binder.jaas.loginModule‘是一个未知的属性。
下面给出了我的项目.pom文件
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
浏览 5提问于2017-10-24得票数 1
我有以下docker-compose.yml我有数据流服务器运行，kafka，zookeeper，mysql，指标收集器。
配置：(我只提供数据流-服务器(1.5.2版本)，指标收集器)指标收集器:镜像: spring.cloud.stream.kafka.binder.brokers=kafka:9092 / metrics - spring.cloud.stream.kafka.binder.zkNodes=zookeeper:2181 -kafka-10环境：- spring.security.user.name=spring - spring.security.user.passwor
浏览 0提问于2018-10-02得票数 0
我正在使用spring cloud kafka绑定器将数据读取到KStream。在阅读其中一个主题的数据时，我需要从头开始阅读。
我已经尝试设置kafka偏移重置和开始偏移属性。但是，找不到任何参考资料。
你能帮我提供一些示例application.yaml来重置偏移量吗，这样我就可以从一开始就消费主题中的消息
添加我使用过的application.yaml：
spring.cloud.stream.bindings.input:
destination: input-topic1
consumer:
useNativeDecoding: true
headerMode
浏览 11提问于2018-05-04得票数 1
我试图使用在Kafka流上运行交互式查询。我坚持从InteractiveQueryService检索密钥值存储。在我的代码上，甚至在示例代码上，我总是会遇到相同的错误：
java.lang.IllegalStateException: Error retrieving state store: prod-id-count-store
at org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService.lambda$getQueryableStore$0(InteractiveQuerySer
浏览 18提问于2022-10-13得票数 0
回答已采纳
我正在使用spring云流并试图发布avro消息，但是得到了上述异常，
我有下面的房产，
spring.cloud.stream.bindings.feed-output-channel.producer.useNativeEncoding=true
spring.cloud.stream.bindings.feed-output-channel.destination=TOPI.NAME
spring.cloud.stream.bindings.feed-output-channel.producer.partition-count=1
spring.cloud.stream.binding
浏览 14提问于2022-01-21得票数 0
回答已采纳
我正在运行JDK 11、Spring 2.5.5和Spring 2020.0.0 5和Kafka 2.7.2。
当我有带有业务对象的消息类型的这个使用者代码时：
@Bean
public Consumer<Message<MyObjectType>> process() {
return input -> {
logger.info("Message = {}", input.getPayload());
};
}
我得到
浏览 7提问于2022-01-17得票数 0
1回答
我正在尝试使用spring应用程序中的kafka主题。我使用的是Spring云流和下面提到的版本
Spring-boot-starter-parent: 2.5.7春季云版本: 2020.0.4
下面是代码和配置
application.yml
spring:
zipkin:
sender:
type: kafka
kafka:
bootstrap-servers:
- localhost:19091
cloud:
stream:
bindings:
audit-in-0:
destinat
浏览 4提问于2021-11-29得票数 2
pom.xml
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.c
浏览 2提问于2018-04-13得票数 0
回答已采纳
当@StreamListener处理失败，Spring Cloud Stream Kafka绑定器尝试将消息重新路由到DLQ时，我得到以下异常。使用Spring Cloud Edgware.SR5。 org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'my.message.destination.my.message.group.errors'; nested exception is java.lang.ClassCastException: j
浏览 30提问于2019-03-23得票数 0
回答已采纳
1回答
我有一个使用SpringBoot2.0.3的Spring应用程序。在Finchley.SR1下，libraries和Spring库。具有以下特性：
spring:
kafka:
ssl:
protocol: SASL_SSL
keystore-location: classpath:kafka.p12 // created from a PEM with private key and certificate
keystore-password: passw0rd
key-store-type: PKCS12
cloud:
浏览 1提问于2018-08-29得票数 0
回答已采纳
2回答
我正试着用码头启动SCDF。它很好地启动，但是当尝试部署任何流时，我会在应用程序日志中得到以下错误
2017-03-09 15:19:45.864 INFO 62 --- [ main] kafka.utils.VerifiableProperties : Verifying properties
2017-03-09 15:19:45.864 INFO 62 --- [ main] kafka.utils.VerifiableProperties : Property client.id is overridden
浏览 4提问于2017-03-09得票数 1
回答已采纳
1回答
我正在尝试设置一个简单的云流Sink，但始终会遇到以下错误。
我试过几个绑定器，它们都给出了相同的错误。
"SEVERE","logNameSource":"org.springframework.boot.diagnostics.LoggingFailureAnalysisReporter","message":"
***************************
APPLICATION FAILED TO START
***************************
Description:
Parame
浏览 0提问于2018-02-14得票数 3
1回答
当前，如果我使用依赖项启动一个服务(下面)，而没有运行，它将失败。
我想把这个改变为做两件事：
启动服务，但等待动物园管理员可用
允许指定一个特定的kafka来代替它，如果它自己旋转它自己。
我知道错误：
org.springframework.context.ApplicationContextException: Failed to start bean 'outputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException:
浏览 5提问于2017-08-25得票数 0
1回答
作为测试的一部分运行了一个嵌入式kafka实例。我正在尝试验证所有消息是否都已被读取，但从kafka管理客户端获得的结果为空。 Map<TopicPartition, OffsetAndMetadata> partitionOffset = embeddedKafkaRule.getEmbeddedKafka().doWithAdminFunction(admin -> {
try{
return admin.listConsumerGroupOffsets(COUNTER_GROUP).partitionsToOffsetAndMe
浏览 51提问于2020-12-17得票数 0
我在spring-云流-schema- 1.2.0.RC1中缺少了AvroSchemaMessageConverter类。我想通过这个链接。
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
<version>1.2.0.RC1</version>
</dependency>
当我查看spring-c
浏览 4提问于2017-04-03得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
