分支到多个主题时的 Kafka Streams 架构问题答案

【问题标题】：Kafka Streams Schema Issue when Branching to Multiple Topics分支到多个主题时的 Kafka Streams 架构问题
【发布时间】：2019-10-25 16:25:10
【问题描述】：

我有一个应用程序，它根据一些规则分支到 20 个主题之一。它因以下错误而开始失败。

stream-thread [pbel-topic-splitter-bc7189fd-e35c-454d-b5e7-97d5e514dc36-StreamThread-1] Failed to process stream task 0_14 due to the following error:
org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_14, processor=KSTREAM-SOURCE-0000000000, topic=com.bns.prm. raw.DETAIL, partition=14, offset=334
        at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:367)
        at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:104)
        at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:413)
        at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:862)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)
Caused by: org.apache.kafka.common.errors.SerializationException: Error registering Avro schema: {"type":"record","name":"DETAIL","namespace":"com.        bns.eventlayer.prm.avro","fields":[{"name":"isb_IDENTITY_NUMBER","type":"long"},{"name":"ND_TRAN_ID","type":["null","long"],"default":null},{"name"        :"D nnect.data.Timestamp","logicalType":"timestamp-millis"}}],"connect.name":"com.bns.eventlayer.prm.avro.DETAIL"}
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema being registered is incompatible with an earlier schema; error code: 409
        at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:209)

我比较了这两种模式，发现了这种细微的差别，不确定这是否重要...

之前

{
         "name":"SD_ACH_BATCH_OFFSET_TYP",
         "type":[
            "null",
            "string"
         ],
         "default":null
      }

之后

      {
         "name":"SD_SECURITY_LEVEL",
         "type":[
            "null",
            {
               "type":"string",
               "avro.java.string":"String"
            }
         ],
         "default":null
      },

【问题讨论】：

标签： apache-kafka avro apache-kafka-streams confluent-schema-registry

【解决方案1】：

这是我的错，我必须清除 错误主题 上的旧架构。写了一个程序来寻找总是崩溃的偏移量（DLQ），读取DLQ的数据并弄清楚它的错误主题......

【讨论】：