无法使用Flink从Kafka检索正确的消息

Question

我是Flink的新手，今天我遇到了一个奇怪的情况。

我运行Kafka服务器，然后使用confluent producer发送消息。使用consumer，我得到了正确的信息，但在应用程序中，我不能。我使用此图像设置message brokerconfluentinc/cp kafka:5.4.1

我用这个向Kafka服务器发送消息

./kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic test \
--property 
value.schema='{"type":"record","namespace":"com.example.kafka","name":"User","fields": 
[{"name":"name","type":"string"},{"name":"age","type":"int"}]}'

我发送的消息是{"name"："Huy"，"age"： 12}

我用这个来听Kafka的留言

./kafka-avro-console-consumer --topic test \
--bootstrap-server localhost:9092

这是我的密码

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
KafkaSource<String> source  = KafkaSource.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("test")
            .setGroupId("console-consumer-29300")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
    env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka").print();
env.execute();

当我将kafka源代码更改为KafkaSource时

user class是由avro-maven-plugin从avro模式生成的类，我在命令中将消息发送到kafka。

你们对此有什么想法吗？一个有效的示例代码确实有很大帮助。我找不到任何最新的样品。谢谢你。

Answer 1

SimpleStringSchema无法读取Avro，因此您的终端正在输出解码的UTF-8。

AvroDeserializationSchema可能会解析字节，但不知道注册表中的模式是否正确。

ConfluentRegistryAvroDeserializationSchema是您想要的。例如，看看测试代码

Answer 2

最后，我需要做的将是像这样使用ConFluentCORstryAvroDeseriazationSchema

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    KafkaSource<User> source  = KafkaSource.<User>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("test")
            .setGroupId("console-consumer-29300")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(ConfluentRegistryAvroDeserializationSchema.forSpecific(User.class, "http://localhost:8081"))
            .build();
    env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka").print();
    env.execute();

我需要指定指向模式注册表的链接以及从avro模式生成的类。如果没有指向模式注册表的url，它将无法工作

谢谢@OneCricketeer的指导。