节点：向Kafka错误发送原型消息

Question

我正在尝试使用HDFS kafka连接器将protobuf消息从kafka发送到HDFS。我的连接器配置如下所示

{
    "name": "hdfs3-connector-test",
    "config": {
        "connector.class": "io.confluent.connect.hdfs3.Hdfs3SinkConnector",
        "tasks.max": "1",
        "topics": "test-topic",
        "hdfs.url": "hdfs://10.8.0.1:9000",
        "flush.size": "3",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
        "value.converter.schema.registry.url":"http://10.8.0.1:8081",
        "confluent.topic.bootstrap.servers": "10.8.0.1:9092",
        "confluent.topic.replication.factor": "1"
    }
}

为了测试这一点，我尝试在一个小节点应用程序中发送protobuf序列化消息。这是我的文件：

// data.proto

syntax = "proto3";
package awesomepackage;

message SearchRequest {
  string query = 1;
  int32 page = 2;
}

和我的节点应用程序

const { Kafka } = require('kafkajs')
const protobuf = require('protobufjs')

const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['10.8.0.1:9092']
})

const producer = kafka.producer()


const run = async () => {
    await producer.connect()

    protobuf.load('data.proto', async (err, root) => {
        console.log("TESTING")
        console.log(err)
    
        let SearchRequest = root.lookupType('awesomepackage.SearchRequest')
        let payload = {query: "test", page: 2} 
    
        var errMsg = SearchRequest.verify(payload);
        console.log(errMsg)
    
        let msg = SearchRequest.create(payload)
        var buffer = SearchRequest.encode(msg).finish();
        console.log(buffer)
        await producer.send({
            topic: 'test-topic',
            messages: [
                {key: 'key1', value: buffer}
            ]
        })
    })
    
}

run()

但是，当我运行此程序时，我会出现以下错误：

Failed to deserialize data for topic test-topic to Protobuf

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Protobuf message for id -1

Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

我该如何解决这个问题？我的猜测是，我的protobuf模式没有在Kafka模式注册表中注册，但我不确定。如果是这种情况，是否有方法将模式从节点发送到注册表？

Answer 1

io.confluent.connect.protobuf。ProtobufConverter需要模式注册表，而不是纯序列化Protobuf。换句话说，节点代码中缺少模式注册表部分（或手动创建“包装”的原型消息的字节）

参考线材格式-合流

如果你不想使用Schema注册表，你可以使用BlueApron原型转换器，但看起来你正在使用一个，所以最好使用合流转换器