有没有一种方法可以在没有魔法字节的情况下使用Kafka模式注册表？

Question

我试图使用confluent的模式注册表使我的应用程序工作，但此时我并不是生产者的完全控制者，你甚至可以将它们视为不绑定confluent产品的遗留应用程序。

我正在查看融合信息，似乎所有消息都应该在有效负载中包含一个魔法字节和模式ID
https://docs.confluent.io/3.2.0/schema-registry/docs/serializer-formatter.html

或者，当我尝试使用它时，我会出现一个错误：

[2020-09-25 13:12:09,008] ERROR WorkerSinkTask{id=s3_parquet_connector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
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.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:491)
            at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)
            at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:324)
            at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:228)
            at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:200)
            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: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic com.obj_pos to Protobuf: 
            at io.confluent.connect.protobuf.ProtobufConverter.toConnectData(ProtobufConverter.java:123)
            at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
            at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:491)
            at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
            at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
            ... 13 more
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!
[2020-09-25 13:12:09,010] ERROR WorkerSinkTask{id=s3_parquet_connector-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask)

我的问题是，是否有一种方法可以禁用这个神奇的字节检查，或者我可以创建一个Kafka流，只在初始消息中附加一个this 5字节，这样之后我就可以与连接到模式注册表的消费者一起使用它。

发生的事情是生产者超出了我的控制，所以我需要以某种方式能够反序列化不包含这5个字节的消息，因为它们是由不依赖于汇合序列化器/反序列化器的生产者产生的

Answer 1

它们由不依赖合流连载器的制作人制作

那么问题不在注册处。

您不应该使用Confluent编写的转换器来使用这些消息，因为这些消息绑定到注册表，并且没有办法跳过它。

您可以使用BlueAporter（假设数据真的是protobuf），或者编写自己的转换器类。