使用kafka创建Spring云流的单元测试时出错
Kafka是一个分布式流处理平台，它具有高吞吐量、可扩展性和容错性的特点。Spring Cloud Stream是一个用于构建消息驱动的微服务的框架，它提供了与消息中间件集成的能力。在使用Kafka创建Spring Cloud Stream的单元测试时出错，可能是由于以下原因导致的：
- 依赖配置错误：首先，需要在项目的依赖管理中添加Spring Cloud Stream和Kafka的相关依赖。确保版本兼容，并正确配置相关的属性，如Kafka的地址、主题等。
- 单元测试环境配置错误：在单元测试中，需要模拟Kafka的环境。可以使用EmbeddedKafka来创建一个嵌入式的Kafka服务器，以便在测试中使用。确保正确配置EmbeddedKafka的相关属性，如端口号、主题等。
- 测试代码编写错误：在编写单元测试代码时，需要确保正确使用Spring Cloud Stream和Kafka的相关注解和API。例如，使用
@EnableBinding注解来绑定消息通道，使用@StreamListener注解来监听消息等。同时，需要模拟发送和接收消息的场景，并进行相应的断言和验证。
相关·内容
我的理解是，创建是为了与Kafka客户端API交互，后来创建项目是为了“构建与共享消息传递系统相连接的高度可伸缩的事件驱动的微服务”，该项目包括两个绑定程序，其中一个是允许与Kafka交互的绑定器。
spring-cloud-stream-binder-kafka-streams
因此，我很清楚，如果我想与Kafka交互，我将使用spring云流方法和适当的绑定器。
但是，我发现您可以与交互，也可以使用spring方法。
需要以下两个依赖项。就是一个例子。
'org.springframework.kafka:spring-kafka'
'org.apache.kafk
浏览 3提问于2021-11-10得票数 0
回答已采纳
我们正在开发SCDF上的项目，kafka版本2.4.1用于数据流，2.3.1用于skipper。我有一个云流应用程序，它在我的IDE中进行单元测试时工作得很好。但当我注册为处理器时，我在我的Stream中看不到此特定应用程序的日志中的任何活动。 当前使用Boot2.2.5和Hoxton.SR3。 但当我使用spring boot版本2.1.4时，这款应用运行得很好。我使用spring boot 2.1.4的原因是这个URL上的示例应用程序使用2.1.4 因此，如果有文档可以让您知道SCDF发行版使用哪些spring boot版本，这会很有帮助。
浏览 13提问于2020-03-25得票数 0
2回答
使用Kafka作为微服务体系结构中的消息传递系统，使用spring相对于spring云流+spring-云-初学者流-kafka有什么好处？
spring云流框架支持更多的消息传递系统，因此具有更模块化的设计。但是功能呢？泉-卡夫卡的功能和弹簧云流+弹簧云-启动-流-卡夫卡的功能之间有差距吗？哪个API设计得更好？
期待阅读你的观点
浏览 5提问于2017-11-15得票数 56
我使用的是最新的Greenwich.SR1，包括spring云流(version: Fishtown.SR2)和本地启动的kafka客户端版本:2.2.0，使用kafka_2.12-2.2.0.jar
我想使用最新的kafka客户端2.1或更高版本的spring云流，因为它包含一些重要的bug修复。但是当我运行我的spring应用程序时，它的日志上写着
INFO 37812 --- [main] o.a.kafka.common.utils.AppInfoParser : Kafka version : 2.0.1
如何与最新的kafka客户端一起使用spring云流？我想使用max.task
浏览 0提问于2019-05-21得票数 0
1回答
使用spring云数据流运行我的spring云流应用程序的最简单方法是什么？我的想法是创建自定义的源代码，使用kafka绑定器和接收器作为预构建的启动程序，文件。我正在创建applciation，它非常基本，类似于。我尝试创建可执行的jar，并使用dataflow shell将其注册为源代码，但这根本不起作用。我已经在跟踪这个了。
浏览 2提问于2016-07-27得票数 0
回答已采纳
1回答
我正在从Spring迁移到Spring数据流。当我寻找模块列表时，我意识到有些源没有在中列出--其中之一就是KAFKA源代码。
我的问题是为什么在春云数据流中从标准源列表中删除KAFKA源？
浏览 4提问于2016-07-24得票数 4
回答已采纳
1回答
我正在使用IBM构建一个应用程序。所建议的构建它的架构是使用spring云流。我在许多文章中读到，您可以将spring云消息传递系统移植到其他系统。这意味着，我以后也可以将IBM更改为kafka，但可以在类路径上提供绑定器实现。那是什么意思？
我使用rabbitmq构建了一个Spring应用程序，我在应用程序中使用的特性是belo，
请求答复
Dlq
生产者和监听器配置都是向上和向下的，这意味着我可以创建一个队列，它的交换和绑定路由密钥使用代码。
如果我使用spring云流，我想知道上面的特性是否可用，是否需要将实现代码更改为将ibm移植到kafka？
浏览 1提问于2018-11-25得票数 0
我正在尝试将春云流与春云功能webflux集成在一起。
由于它们在将来的版本中不推荐spring云反应性流，所以我尝试使用spring云函数。
Spring云web函数可以公开其函数的端点，路径类似于doc。
从云流中，我可以看到源需要定义为供应商
但是我的用例是从反应性http端点获取POST数据，并将其摄取到kafka中，有没有办法从spring云功能web和spring云流中实现呢？
从文档中获取包含弹簧云流的spring云函数。
@SpringBootApplication
@EnableBinding(Source.class)
public static class Source
浏览 2提问于2019-03-27得票数 2
回答已采纳
我使用过spring cloud data flow、rabbitmq和kafka，但我想知道是否可以使用google pub/sub安装scdf。
我不想创建一个流(新的应用程序，spring云流)，它的源或宿都是gcp，我想让google pub/sub over spring cloud数据流服务器作为一个中间消息传递代理。
有什么建议吗？
浏览 1提问于2019-12-30得票数 1
1回答
我正致力于为银行创建一个大容量的JSON数据处理应用程序，使用Spring、Kafka和QuickFIX/J。这是我第一次使用Kafka和QuickFIX/J这样的技术，无法决定是使用普通的Kafka spring还是Kafka流还是spring云流。以下是要求：
从多个进程读取数据，并将数据发送给QuickFIX/J启动器，后者进一步将其发送到外部修复引擎-- QuickFIX/J受体从外部修复引擎接收数据，并将其再次写入多个卡夫卡主题，但这次不同的是。
我读过一些教程/文章，这些文章说Kafka流或spring云流是好的，如果您有消费者/生产者，执行高容量的数据流，并希望实现一次处理。但是
浏览 7提问于2020-06-01得票数 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
回答已采纳
最近发现了CVE-2022-22963弹簧云函数漏洞，这里的团队提供了一个简单的修补程序：。
我们通过以下依赖关系从spring-cloud-function-context:3.0.6获得spring-cloud-stream:3.0.4 ( spring.boot.version为2.2.6)：
spring-cloud-services-starter-circuit-breaker:2.2.6
弹簧云总线:2.2.6(包括amqp/kafka)
弹簧-云-启动-流-卡夫卡:3.0.4
春-云-启动-流-兔:3.0.4
我们没有直接使用RoutingFunctio
浏览 20提问于2022-04-12得票数 1
回答已采纳
我正在使用spring云流kafka绑定器和模式注册表。(不是kakfa流)我要做的是，当不可反序列化消息进入输入主题时，将deserializable消息发送到dlq.。
所以我尝试了像下面这样的配置，但是spring应用程序让不断地无限地重试，并且说
由: org.apache.kafka.common.errors.SerializationException:错误反序列化id -1的Avro消息引起的
spring.cloud.stream:
function:
definition: functionName
bindings:
functionName-i
浏览 3提问于2021-02-18得票数 0
回答已采纳
我想了解一下spring云数据流的体系结构。特别是使用它来编排纯粹的批处理管道。我已经在k8s中部署了所有组件，并且注意到我可以在没有中间件(Kafka / RabbitMQ)的情况下部署skipper组件，并通过服务器组件成功地运行批处理作业。
根据我可以看到的文档，skipper用于编排流作业。当纯粹用于批处理作业或任务时，skipper (以及随后的Kafka/RabbitMQ中间件)是否是春季云数据流的必要部分？
浏览 0提问于2020-04-23得票数 0
回答已采纳
1回答
在功能编程模型中使用了kafka和spring云流。我想使用反应性api
因此，我有一个函数bean，它接受Flux并返回通量，返回的通量是在单独的类中创建的。
我需要订阅来激活新的/返回的流量吗？
浏览 4提问于2022-01-18得票数 0
我想在我的春季云流卡夫卡应用程序中为我的生产者设置acks=all属性。
我试过这样做：
spring.cloud.stream.kafka.binder.requiredAcks=all
和
spring.cloud.stream.kafka.streams.binder.configuration=all
和
spring.cloud.stream.kafka.streams.bindings.<channel>.producer.configuration.requiredAcks=all
不幸的是，注意到对我有用。
请您帮助我如何将这些属性设置为应用程序级别或生产者/使用者
浏览 1提问于2019-04-24得票数 1
1回答
春云流不能产生卡夫卡讯息
、、、
我试图使用spring云流向Kafka发送消息，但是send方法会产生以下错误：
Caused by: java.lang.IllegalArgumentException: Magic v1 does not support record headers
at org.apache.kafka.common.record.MemoryRecordsBuilder.appendWithOffset(MemoryRecordsBuilder.java:403) ~[kafka-clients-1.0.0.jar:na]
...
App.java
package com.example.demos
浏览 0提问于2018-03-20得票数 0
回答已采纳
1回答
我们已经创建了kafka主题，默认replication-factor为1，为了更新这一点，我们在application.yaml中添加了spring.cloud.stream.kafka.binder.replication-factor: 3属性，该属性没有反映在kafka中创建的主题。 使用的spring云流版本为Germantown.SR1
浏览 17提问于2020-01-27得票数 2
回答已采纳
1回答
我正在使用弹簧云流卡夫卡同步生产者在一个弹簧引导微服务。每次我们部署该服务时，对kafka的第一次调用就需要超过20秒才能将消息发布到主题。但随后的所有调用几乎不需要3到4毫秒。此问题也是随机发生的，是间歇性的，但大多数发生在我们重新启动服务时。我们使用kafka版本0.9.0.1和gradle依赖项，如下所示
compile('org.springframework.cloud:spring-cloud-starter-stream-kafka')
}
dependencyManagement {
imports {
mavenBom "org.springframe
浏览 2提问于2017-05-25得票数 1
我遇到了一个问题，我在spring中有一个应用程序，这个应用程序从一个主题中消耗，在另一个主题中生成。
应用程序使用的主题是现场，应用程序生成的主题是在云aws上。
是否有方法为每个主题指定引导服务器和模式注册表？
我的application.properties具有以下属性：
spring.kafka.bootstrap-servers=localhost:32202 spring.kafka.properties.schema.registry.url=127.0.0.1:8082
这里的问题是，这些属性是为消费者和生产者。我需要为使用者指定一个引导服务器，为生产者指定另一个引导服务器。还
浏览 5提问于2022-11-08得票数 0
当将事件发布到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
回答已采纳
我正在使用KStreams和Avro创建一个应用程序。我曾尝试将嵌入式kafka用于单元测试用例，但是得到了以下异常：
创建名为“kafkaEmbedded”的bean时出错:调用init方法失败；嵌套异常为java.lang.NoSuchMethodError: java.lang.NoSuchMethodError
有没有更好的方法/解决方案来用kafka流编写单元测试用例？
浏览 1提问于2018-04-09得票数 2
回答已采纳
我有一个春季引导应用程序，在那里我使用apache streams。我不使用春云流。我添加了执行器健康检查端点。我在application.yml中这样配置了它：
management:
health.db.enabled: false
endpoints.web:
base-path:
path-mapping.health: /
当一个运行时异常被抛出并且我的流被停止时，日志显示，但是健康检查状态已经启动。
2019-09-17 13:16:31.522 INFO 1 --- [ Thread-5] org.apache.kafka.streams.KafkaStr
浏览 13提问于2019-09-19得票数 7
回答已采纳
1回答
我正在使用与kafka经纪公司的春云流进行微服务互传。作为其中的一部分，stream bridge将用于发送消息，这是很好的。
--但是在使用所述消息期间，不需要立即使用该消息，而是当满足条件时，只需要使用。
从文档中，我了解到我需要在这方面使用Polled Consumers(如果我弄错了就纠正我)。
这就是我对的理解所做的尝试。
application.properties
spring.cloud.stream.pollable-source = consumeResponse
spring.cloud.stream.function.definition = consumeRespon
浏览 7提问于2022-03-10得票数 0
回答已采纳
我们正在使用Kafka绑定构建带有Spring云流的定制DB Sink。我们使用以下版本的Spring和Spring
<spring-cloud.version>Hoxton.SR11</spring-cloud.version>
<spring.boot.version>2.3.4.RELEASE</spring.boot.version>
春云流卡夫卡绑定依赖-
<dependency>
<groupId>org.springframework.cloud</groupId>
<art
浏览 2提问于2021-04-30得票数 0
我们从输入主题中读取拓扑(使用binder: x地址: x)，并使用spring云流kafka流处理记录并将其写入输出主题(使用binder: y地址: y)。记录不会写入输出主题。但是，当我将绑定器(代理地址)设置为相同(x或y)时，记录将被写入主题y。是否应该在拓扑结构中使用相同的代理？我需要使用不同的绑定程序和代理作为输入和输出主题？我怎么才能解决这个问题？
错误: 2021-06-17 12:17:21.483警告20848 --读取-1-生产者clientId=inputTopic-32100000000000000000015-f0bd5423-e670-43e8-ab0b-84e
浏览 4提问于2021-06-17得票数 0
回答已采纳
1回答
我试图使用函数编程(和spring云流)从输入主题转换输入AVRO消息，并在输出主题上发布新消息。这是我的转换函数：
@Bean
public Function<KStream<String, Data>, KStream<String, Double>> evenNumberSquareProcessor() {
return kStream -> kStream.transform(() -> new CustomProcessor(STORE_NAME), STORE_NAME);
}
CustomProcessor是一个实现“转换
浏览 5提问于2021-04-06得票数 1
在春季引导应用程序中，我正在使用spring云启动器-流- kafka和spring-云流罐子。
jar的版本是1.1.1.RELEASE，我们正在创建名为Test的生产者。对于同样的情况，下面是application.yml文件。
spring.boot.cloud.stream.bindings.output:
destination: Test
content-type: application/octet-stream
浏览 3提问于2017-11-14得票数 1
回答已采纳
得到：
创建名为“embeddedKafka”的bean时出错
原因: java.lang.ClassNotFoundException: com.codahale.metrics.Reservoir
<kafka.version>2.8.1</kafka.version>
<spring.kafka.version>2.8.1</spring.kafka.version>
springboot版本: 2.6.3 -不能更改
<dependency>
<groupId>org.s
浏览 17提问于2022-07-25得票数 0
回答已采纳
我试图将目前用Java代码编写的spring应用程序的配置具体化。是将ProducerConfig和ConsumerConfig值放入spring.kafka.streams.properties，还是通过spring.kafka.producer和spring.kafka.consumer正确配置它们？
到目前为止，看来我应该把我所有的配置都放到KafkaStreamsConfiguration类型的bean中，以便配置我的kafka-streams应用程序。目前，我是通过直接在代码中设置ProducerConfig和ConsumerConfig值来做到这一点的。
当我将此配置具体化时，从P
浏览 1提问于2019-06-13得票数 3
回答已采纳
1回答
支持嵌入式头吗？
、、、、
根据这一主题：
-这不是卡夫卡的头支持
但是说：
spring.cloud.stream.kafka.binder.headers
将由绑定程序传输的自定义标头列表。
默认值:空。
I无法让它与弹簧云流绑定-kafka:1.2.0
发送日志：
MESSAGE (e23885fd-ffd9-42dc-ebe3-5a78467fee1f) SENT :
GenericMessage [payload=...,
headers={
content-type=application/json,
correlationId=51dd90b1-76e6-4b8d-b667-da
浏览 3提问于2017-06-15得票数 3
回答已采纳
由于不触发SleuthKafkaAspect.wrapProducerFactory()方法，跟踪信息不会在kafka消息上传播。在生产者端，消息被正确发送，跟踪信息被正确记录。在消费者方面，反而创建了一个新的traceId和spanId。
以下两条日志记录行显示了traceId、spanId (和parentId)的不同值：
2021-03-23 11:42:30.158 [http-nio-9185-exec-2] INFO my.company.Producer - /4afe07273872918b/4afe07273872918b// - Sending event='My
浏览 4提问于2021-03-23得票数 0
回答已采纳
2回答
它似乎是非常简单的实现，只有两个特定于云流项目的库，但我得到了
java.lang.ClassNotFoundException: org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter
项目中的所有依赖项都是：
compile(
"org.springframework.boot:spring-boot-starter-web",
"org.springframework.boot:spring-boot-starter-actua
浏览 3提问于2017-09-28得票数 0
回答已采纳
以下是我迄今为止在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
回答已采纳
4回答
我有一个问题，反序列化来自卡夫卡主题的信息。这些消息已经使用spring云流和Apache序列化。我正在使用Spring阅读它们，并试图对它们进行反序列化。如果我同时使用spring来生成和使用消息，那么我可以很好地反序列化这些消息。问题是当我用Spring消费它们，然后尝试反序列化。
我使用的是模式注册表(用于开发的spring架构注册表，以及生产中的合流模式)，但是反序列化问题似乎发生在调用Schema的事件之前。
很难将所有相关代码都发布在这个问题上，所以我已经将其发布在git中心的回购程序中：。
我在这个主题上发送的对象只是一个简单的pojo：
@Data
public class R
浏览 0提问于2019-01-30得票数 3
回答已采纳
我使用SpringCloud2020.0.2和spring :2.6.6和SpringBoot2.5.8。这个很好用。但是当我将spring云升级到2020.0.4时，spring就会抛出以下错误。
MessageHandling异常，缺少标头kafka_acknowledgement。
如果弹簧云2020.0.2和spring :2.6.6兼容，谁能帮我吗？谢谢
浏览 4提问于2022-02-14得票数 0
回答已采纳
我使用弹簧云流:3.2.2弹簧云流绑定卡夫卡:3.2.5 spring-cloud-stream-binder-kafka-streams:3.2.5
我想在反应性编程中写出消费者卡夫卡。我使用avro模式注册表。在我的例子中，我在一个主题中输入了多个事件。我的使用者使用所有类型，但我想在每个事件类型中编写一个使用者。在您的文档中，我找到了一些关于路由的信息。在反应模式下，我只能在application.yml中使用路由表达式。但这对我没用。你能帮我吗？我试过几件事，但我不明白为什么它不起作用。我的2消费者消费所有的事件类型，不是特定的。
我的两个消费者：
@Bean
public C
浏览 10提问于2022-09-30得票数 0
在Spring项目中，这些maven依赖项之间有什么区别-
spring-cloud-stream-binder-kafka和spring-cloud-starter-stream-kafka
浏览 2提问于2020-07-22得票数 1
回答已采纳
这不是一个大问题，但我很好奇，一些额外的流消费者来自哪里，如果这是一个设置，我可以改变。
我有一个非常简单的spring云流消费者设置，针对当地的Kafka代理。这是spring的配置
spring:
cloud:
stream:
bindings:
consumer-in-0:
destination: test-topic
group: test-group
而消费者阶层本身：
@Bean
Consumer<Message<String>> consumer() {
return me
浏览 7提问于2021-04-14得票数 0
回答已采纳
1回答
是否有任何支持或计划支持avro和/或模式注册中心的汇合平台在春季云流？我看到在春季集成中存在着对avro的依赖-kafka 1.3.0，dependency被排除在spring云流绑定卡夫卡之外，但是spring-integration-kafka (2.0)和spring的主分支没有我能找到的任何avro依赖。
浏览 5提问于2016-05-20得票数 2
回答已采纳
2回答
我有一个应用程序，我想要在卡夫卡消息中手动执行(n)ack。根据spring云文档，它应该使用autoCommitOffset 来完成。
但是，在我的应用程序中，即使定义了这样的属性，头KafkaHeaders.ACKNOWLEDGMENT仍然是空的。
下面是我的配置
spring.cloud.stream.kafka.binder.brokers=${KAFKA_BROKER_LIST}
spring.cloud.stream.default.contentType=application/json
spring.cloud.stream.bindings.mytopic.destinati
浏览 7提问于2020-05-14得票数 1
回答已采纳
1回答
Kafka用例替换控制M作业
、、、、
我几乎控制不了生产中的M个作业。
第一项任务-加载csv文件记录并插入数据库暂存表第二项任务-对临时表第三作业中的每个记录执行数据封装-从暂存表中读取数据并插入到另一个表。
目前，我们使用apache骆驼来完成这个任务。
我们有合流卡夫卡许可证，所以我们想使用卡夫卡。
设计方案
创建csv kafka源连接器来读取csv中的数据，并插入kafka输入主题创建春云kafka流绑定应用程序，从输入主题读取数据，丰富数据并将数据推送到输出主题，让kafka接收器连接器将数据从输出主题推送到数据库。
现在问题二，我们需要有数据库连接来丰富数据，当在youtube上观看视频时，它说春云卡夫卡流绑定器不应
浏览 1提问于2021-05-14得票数 0
我在这个网站上发布了基于的Spring云数据流。
我创建了3个应用程序，源代码，处理器和水槽。
我在Spring数据流云仪表板上注册了它们。
然后，我创建了一个流，其中源连接到处理器，连接到接收器。
当我部署应用程序并打开{流名}时，
我得到以下错误，
{
logs: {
testdir-stream.sender1-v3: "Error: Unable to access jarfile /testdir/data-sender-0.0.1-SNAPSHOT.jar ",
testdir-stream.sink1-v3: "Error: Unable to acce
浏览 0提问于2019-09-09得票数 0
回答已采纳
我正在尝试开发一个新的应用程序来在SCDF 2.4.1和Skipper 2.3.1上工作 我从https://github.com/sabbyanandan/stream-programming-models那里取了样本 我在本地构建的。下载了SCDF kafka的docker compose，设置版本并挂载我的repo，然后开始我的docker compose。 当我部署"function“模块并创建一个简单的流http | customUpper | log时 我看到示例运行良好，并且能够像预期的那样看到日志输出。 当我修改函数流app时，使用Spring Boot、2.2.4和
浏览 25提问于2020-03-26得票数 0
无法在Spring Cloud Stream Binder Kafka 3.x (functional style)中将自定义商店连接到我的Transformer，如here中的示例所示。 我将KeyValueStore定义为具有StoreBuilder<KeyValueStore<String,Long>>类型的@Bean @Bean
public StoreBuilder<KeyValueStore<String,Long>> myStore() {
return Stores.keyValueStoreBuilder(
浏览 31提问于2021-04-06得票数 0
回答已采纳
1回答
我使用的是spring 2.0.1，但是当我执行测试时，它似乎是在测试类之前开始的，也是主类。在我的主要课程中，我使用spring云配置、发现服务和kafka。我有一门考试课：
@RunWith(SpringRunner.class)
@SpringBootTest
@TestPropertySource(locations = "classpath:bootstrap-test.properties")
@ActiveProfiles("test")
public class DaemonLogServiceTest {
@Autowired
private
浏览 0提问于2018-06-19得票数 0
回答已采纳
我有一个@StreamListener方法，它将执行REST调用。当REST调用返回异常时，@StreamListener方法抛出RunTimeException并重试。@StreamListener方法抛出RuntimeException时，重试次数不限
Spring云流重试配置：
spring.cloud.stream.kafka.bindings.inputChannel.consumer.enableDlq=true
spring.cloud.stream.bindings.inputChannel.consumer.maxAttempts=3
spring.cloud.stream.
浏览 0提问于2018-08-27得票数 1
我正在尝试构建一个Spring应用程序，它使用来自一个Kafka主题的事件，对其进行一些处理(可能涉及对反应性db查找的操作)，并将它们发送到另一个kafka主题。目前，该应用程序为kafka使用了webflux和spring云数据流。
请有人建议使用好吗，还是应该使用其他方法来实现端到端的反应行为。
尝试用Spring引导WebFlux，春季云数据流卡夫卡流。
这里的处理器示例
浏览 7提问于2022-12-02得票数 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
回答已采纳
我正在试用使用spring云流的kafka流交互查询。
我将码头映像部署到kubernetes，它工作得很好。
然后，我做了一个更改，比如添加一个测试并部署一个新映像，突然它就停止工作了，因为主机信息已经不可用了。HostInfo{主机=“不可用”，端口=-1}
如果我再次从步骤1重新部署映像，那么它将重新开始工作。
通过查看Spring代码，如果集群元数据为空，则在StreamsMetadataState.getMetadataWithKey()中就会出现这种情况。Javadocs指出，这是由于“如果流正在(重新)初始化”。
--我似乎找不到在这种情况下应该做什么的答案，
浏览 2提问于2019-09-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
