我正在尝试为一个定制的流处理器编写单元测试，但在序列化我需要为测试发送的消息时遇到了麻烦。我遵循了卡夫卡的例子：。我为我的流中的一个自定义类(自动生成的avro类)使用了SpecificAvroSerde，但是我不能用MockSchemaRegistryClient()在测试中配置它，我只能指向SR的URL。Serde<MyCustomObject>
浏览 3提问于2018-07-20得票数 0
回答已采纳
我正在使用spring 2.6.3 EmbeddedKafka和Junit 5编写一个spring引导集成测试，用于使用avro消息的拓扑结构。在测试中我使用的是MockSchemaReigstryClient at io.confluent.kafka</
浏览 9提问于2021-01-10得票数 3
我有一个应用程序，当使用转换器执行去重复时，它在处理器API中使用StateStore。
这是因为KafkaStreamsStateStore的默认值serde是一个StringSerde。因此，我在valueSerde注释中添加了KafkaStreamsStateStore参数，指示SpecificAvroSerde的参数：
浏览 0提问于2019-06-17得票数 3
我使用的是带有卡夫卡绑定的小枝云流。我有多个不同内容类型的消费者。是否有一种方法可以配置反序列化器的基础上？
浏览 0提问于2021-03-16得票数 0
回答已采纳
1回答
我很难弄清楚如何测试使用Avro作为消息格式和(合流)模式注册表的应用程序。bindings: consumer: producer:
valueSerde: io.confluent.kaf
浏览 0提问于2019-07-31得票数 1
回答已采纳
1回答
我试图使用函数编程(和spring云流)从输入主题转换输入AVRO消息，并在输出主题上发布新消息。我尝试过许多不同的配置(似乎很难找到正确的文档)，每次应用程序都找不到schema.registry.url的设置。: brokers: my-cluster-kafka-bootstrap.kafka:9092
consumer-propertiesschema.registry.url: http:/&#x
浏览 5提问于2021-04-06得票数 1
我正在尝试构建一个Stream，它获取一个Avro主题，做一个简单的转换，然后以Avro格式再次将其发送回另一个主题，而我有点卡在最后的序列化部分。我已经创建了一个AVRO模式，我正在导入它并使用它来创建特定的Avro Serde。但是我不知道如何使用这个serde将电影对象序列化回AVRO。configure(mapOf(Pair("schema.registry.url", schemaRegistryUrl)), false)
val mov
浏览 15提问于2020-04-05得票数 0
2回答
我正在尝试使用来自kafka主题的合流avro消息，作为带有spring 2.0的Kstream。
我能够将消息作为MessageChannel来使用，而不是以KStream的形式使用。:失败配置值serde类io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde at org.apache.kafka.streams.StreamsConfig.defaul
浏览 2提问于2019-11-25得票数 1
回答已采纳
我的Kafka Stream应用程序有以下配置 config.put(StreamsConfig.APPLICATION_ID_CONFIG); config.put(SpecificAvroSerializer.java:58)
at io.confluent.<em
浏览 0提问于2018-05-15得票数 5
回答已采纳
我试图使用绑定器来使用来自主题的Avro消息，但无法修复这个classCast异常。这是我的代码： public Consumer<KStream<EventKey, Event>> process(){ input.peek<EventKey> avroInSerde = new SpecificAvroSerde<>();
Map<String, Object
浏览 2提问于2021-02-07得票数 0
回答已采纳
对我想要实现的目标做一个简短的解释:我想为avro做卡夫卡流拓扑(使用TopologyTestDriver)的功能测试。问题:无法“模拟”schemaRegistry来自动化模式发布/读取
到目前为止，我尝试的是使用MockSchemaRegistryClient来模拟schemaRegistry，但我不知道如何将它链接到KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http:&
浏览 1提问于2018-10-10得票数 4
1回答
我能否在Kafka流中使用处理器api，并从一个kafka集群中的某个主题读取数据，并在另一个kafka集群中将数据输送到kafka主题中？
使用处理器api是可行的吗？我知道，这是不可能的流DSL！
浏览 0提问于2018-09-16得票数 0
回答已采纳
我需要在KStream中使用transform()操作，但在没有设置必要的serdes时得到通常的ClassNotFoundException： materializedAbsoluteSomeValueFrequency)
其他KStream操作允许我们指定要使用的serdes，但不能在转换的</e
浏览 0提问于2019-08-02得票数 0
2回答
我对Spring kafka Stream是个新手。我最近设置了一个项目，并尝试使用kafka stream API来发布AVRO。需求:需要解析一个复杂的Jsonobject，它可能有多个子jsons。解析Json并为每个子json发送AVRO消息。它也可以是一个或多个。 我知道如何使用kafka客户端API来做到这一点。这很简单，可以用循环来完成，但是用Stream API，我是个新手。当前stream类的代码如
浏览 38提问于2021-08-12得票数 0
1回答
我很难正确地配置，以便Kafka-Streams使用SSL与信任存储和密钥存储。Stream1: Topic1 > Topic2Stream3: Topic4 > Topic5
我使用最新的框架和valueSerde: io.confluent.kafka.streams.serdes.avro.
浏览 5提问于2021-07-15得票数 0
回答已采纳
我正在构建一个电子商务应用程序，其中我目前正在处理两个数据馈送:订单执行和破碎的销售。由于各种原因，失败的销售将是无效的执行。失败的销售将具有与订单相同的订单参考编号，因此连接位于订单参考编号和行项目编号上。
目前，我有两个主题-- orders和broken。两者都是使用Avro模式定义的，并使用SpecificRecord构建。关键是OrderReferenceNumber。= new SpecificAvroSerde<B
浏览 0提问于2018-05-18得票数 2
我正在做一个POC来演示一些Spring Cloud Kafka功能。我使用的是带有SpringBoot 2.4.1的Java11。我的build.gradle具有以下库 implementation 'org.springframework.cloud:spring-cloud-stream'
implemen
浏览 0提问于2021-01-30得票数 0
我正在使用KStreams和Avro创建一个应用程序。我曾尝试将嵌入式kafka用于单元测试用例，但是得到了以下异常：
有没有更好的方法/解决方案来用kafka流编写单元测试用例？
浏览 1提问于2018-04-09得票数 2
回答已采纳
2回答
我有几个Nifi进程组，在升级到生产环境之前，我想对它们运行集成测试。问题是我似乎找不到任何关于如何做到这一点的文档。Data Provenance似乎是一个很有前途的工具，可以实现我想要的功能，然而，在流程文件的生命周期中，数据是向kafka或文件系统发布的，或者从kafka或文件系统发布数据。结果，流文件UUID发生了变化，所以我不能使用nifi-api查询它。此外，我知道Nifi提供了一个TestRunner库
浏览 12提问于2018-08-14得票数 6
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号