avro 相关问题

我正在开发一个流处理应用程序，它连接两个流并将新记录输出到不同的主题。下面是我的配置 @豆公共双功能我正在开发一个流处理应用程序，它连接两个流并将新记录输出到不同的主题。以下是我的配置 @Bean public BiFunction<KStream<String, TestRecord>, KStream<String, TestRecord>, KStream<String, ModifiedTestPayload>> process() { return (walletCreated, placementCreated) -> placementCreated .selectKey((key, val) -> val.getClientId()) .join(walletCreated.selectKey((s, testPayload) -> testPayload.getClientId()), (ts, ts1) -> new ModifiedTestPayload(ts.getClientId()), JoinWindows.ofTimeDifferenceWithNoGrace(Duration.of(2, ChronoUnit.MINUTES)), StreamJoined.with(Serdes.String(), CustomSerdes.TestRecord(), CustomSerdes.TestRecord())); } 两个主题都将输入流以产生相同结构的数据。以下是我定义的自定义 Serdes。 public class CustomSerdes { public static Serde<TestRecord> TestRecord() { return new TestRecordSerde(); } public static class TestRecordSerde extends Serdes.WrapperSerde<TestRecord> { public TestRecordSerde() { super(new TestRecordSerializer(), new TestRecordDeserializer()); } } public static class TestRecordSerializer implements Serializer<TestRecord> { private final KafkaAvroSerializer inner; public TestRecordSerializer() { this.inner = new KafkaAvroSerializer(); } @Override public void configure(Map<String, ?> configs, boolean isKey) { inner.configure(configs, isKey); } @Override public byte[] serialize(String topic, TestRecord data) { return inner.serialize(topic, data); } @Override public void close() { inner.close(); } } public static class TestRecordDeserializer implements Deserializer<TestRecord> { private final KafkaAvroDeserializer inner; public TestRecordDeserializer() { this.inner = new KafkaAvroDeserializer(); } @Override public void configure(Map<String, ?> configs, boolean isKey) { inner.configure(configs, isKey); } @Override public TestRecord deserialize(String topic, byte[] data) { return (TestRecord) inner.deserialize(topic, data); } @Override public void close() { inner.close(); } } public static Serde<ModifiedTestPayload> ModifiedTestPayload() { return new ModifiedTestPayloadSerde(); } public static class ModifiedTestPayloadSerde extends Serdes.WrapperSerde<ModifiedTestPayload> { public ModifiedTestPayloadSerde() { super(new ModifiedTestPayloadSerializer(), new ModifiedTestPayloadDeserializer()); } } public static class ModifiedTestPayloadSerializer implements Serializer<ModifiedTestPayload> { private final KafkaAvroSerializer inner; public ModifiedTestPayloadSerializer() { this.inner = new KafkaAvroSerializer(); } @Override public void configure(Map<String, ?> configs, boolean isKey) { inner.configure(configs, isKey); } @Override public byte[] serialize(String topic, ModifiedTestPayload data) { return inner.serialize(topic, data); } @Override public void close() { inner.close(); } } public static class ModifiedTestPayloadDeserializer implements Deserializer<ModifiedTestPayload> { private final KafkaAvroDeserializer inner; public ModifiedTestPayloadDeserializer() { this.inner = new KafkaAvroDeserializer(); } @Override public void configure(Map<String, ?> configs, boolean isKey) { inner.configure(configs, isKey); } @Override public ModifiedTestPayload deserialize(String topic, byte[] data) { return (ModifiedTestPayload) inner.deserialize(topic, data); } @Override public void close() { inner.close(); } } } 和我的 application.yml 下面 spring: kafka: streams: application-id: test-app cloud: function: definition: process stream: kafka: streams: bindings: process-in-0: consumer: key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde value-serde: io.confluent.kafka.streams.serdes.avro.GenericAvroSerde process-in-1: consumer: key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde value-serde: io.confluent.kafka.streams.serdes.avro.GenericAvroSerde process-out-0: producer: key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde value-serde: ai.wownettests.kf.kafkastreamtestone.CustomSerdes$ModifiedTestPayloadSerde configuration: schema.registry.url: http://localhost:8081 binder: brokers: localhost:9092 configuration: schema.registry.url: http://localhost:8081 specific.avro.reader: true default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde bindings: process-in-0: destination: testkf.placements.created process-in-1: destination: testkf.wallet.created process-out-0: destination: testkf.client.profile_completed logging: level: root: info 当我启动我的应用程序时，我收到如下错误（我有一些关于该应用程序尝试处理的 kafka 主题的数据） org.apache.kafka.streams.errors.StreamsException: Unable to serialize record. ProducerRecord(topic=[test-app-KSTREAM-KEY-SELECT-0000000003-repartition], partition=[null], timestamp=[1706526131230] at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:231) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:175) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:85) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:215) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:154) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:48) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:152) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamTask.lambda$doProcess$1(StreamTask.java:810) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:872) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:810) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:741) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.TaskExecutor.processTask(TaskExecutor.java:97) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.TaskExecutor.process(TaskExecutor.java:78) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1765) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:807) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:617) ~[kafka-streams-3.6.1.jar:na] at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:579) ~[kafka-streams-3.6.1.jar:na] Caused by: org.apache.kafka.common.errors.InvalidConfigurationException: You must configure() before serialize() or use serializer constructor with SchemaRegistryClient 通过创建自定义 serde 的 bean 修复了这个问题 @Bean public Serde<TestRecord> testRecordSerde() { Map<String, ?> config = Collections.singletonMap("schema.registry.url", "http://localhost:8081"); var r = CustomSerdes.TestRecord(); r.configure(config, false); return r; }

apache-kafka spring-kafka avro spring-cloud-stream

回答 1 投票 0

无法在我的 Maven 构建中获取 org.apache.flink.formats 包

我一直在尝试通过maven构建我的apache flink项目，但由于某种原因我遇到了编译错误。值得注意的是“org.apache.flink.formats 包不存在”...

maven apache-flink avro flink-streaming

回答 1 投票 0

可选逻辑类型情况下的 Avro 序列化问题

我没有任何kafka和相关技术的经验，所以我需要你的帮助。我面临以下问题：在我当前的项目中，我们使用融合平台（kafka + schema 注册表）。对于

spring-boot apache-kafka spring-kafka avro confluent-platform

回答 1 投票 0

Debezium 2.0 汇合 AvroConverter 无法工作

自从 debezium 2.0 决定删除 io.confluence.connect.avro.AvroConverter 我尝试直接注入它，但它一直给我以下错误：这是我的简化部署： apiVersi...