我正在使用kafka connect和google pubsub连接器将来自gcp pubsub的消息写入kafka主题。
我的连接器具有以下配置：

{
    "name": "MyTopicSourceConnector",
    "config": {
        "connector.class": "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
        "tasks.max": "10",
        "kafka.topic": "myTopic",
        "cps.project": "my-project-id",
        "cps.subscription": "myTopic-sub",
        "name": "MyTopicSourceConnector",
        "key.converter": "io.confluent.connect.protobuf.ProtobufConverter",
        "key.converter.schema.registry.url": "http://myurl-schema-registry:8081",
        "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
        "value.converter.schema.registry.url": "http://myurl-schema-registry:8081"
    }
}

原始消息值架构如下所示：

syntax = "proto3";

message value_myTopic {
    bytes message = 1;
    string notificationConfig = 2;
    string eventTime = 3;
    string bucketId = 4;
    string payloadFormat = 5;
    string eventType = 6;
    string objectId = 7;
    string objectGeneration = 8;
}

当我使用avro或json（使用适当的转换器）但使用protobuf时，此设置工作正常。我的连接器在部署后立即抛出以下错误消息，但失败：

org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:292)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:321)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:245)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Unsupported root schema of type STRING
        at io.confluent.connect.protobuf.ProtobufData.rawSchemaFromConnectSchema(ProtobufData.java:315)
        at io.confluent.connect.protobuf.ProtobufData.fromConnectSchema(ProtobufData.java:304)
        at io.confluent.connect.protobuf.ProtobufData.fromConnectData(ProtobufData.java:109)
        at io.confluent.connect.protobuf.ProtobufConverter.fromConnectData(ProtobufConverter.java:83)
        at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$1(WorkerSourceTask.java:292)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
        ... 11 more