跟踪oracle数据库更改时无法解析模式注册表接收到的模式

Question

我正在使用confluent和kafka connect oracle(https://github.com/erdemcer/kafka-connect-oracle)为了跟踪Oracle database 11g XE中的更改，我可以使用模式注册表api（如“curl-X GET”）查看模式内容http://localhost:8081/schemas/ids/44" :

{"subject"："Test. KAFKAUSER. TEST-value"，"version"： 1，"id"： 44，"Schema"："{"type"："记录"，"name"："row"，"namesspace"："test.kafkauser.test"，"field"：[{"name"："SCN"，"type"："long"}，{"name"："SEG_OWNER"，"type"："string"}，{"name"："TABLE_NAME"，"type"："string"}，{"name"："TIMESTAMP"，"type"：{"type"："long"，"connect.version"： 1，"connect.name"："org.apache.kafka.connect.data.Timestamp"，"logicalType"："timestamp-millis"}}，{"name"："SQL_REDO"，"type"："string"}，{"name"："OPERATION"，"type"："string"}，{"name"："data"，"type"：["null"，{"type"："记录"，"：["null"，"Double"]，"default"： null}，{"name"："NAME"，"type"：["null"，"string"]，"default"： null}]，"connect.name"："value"}]，"default"： null}，{"name"："前"，"type"：["null"，"value"]，"default"： null}]，"connect.name"："test.kafkauser.test.row"}"，"删除"： false}

但是，此模式无法被conFluent在python中的模式注册表解析：

schemaRegistryClientURL="http://localhost:8081"
from confluent.schemaregistry.client import CachedSchemaRegistryClient
from confluent.schemaregistry.serializers import MessageSerializer
schema_registry_client= CachedSchemaRegistryClient(url=schemaRegistryClientURL)
schema_registry_client.get_by_id(44)

我得到以下错误：

Traceback（最近调用的最后一次）：File"，第1行，在File"build/bdist.linux-x86_64/eges/conFluent/Schemaregistry/Client/CachedSchemaRealstryClient.py"，第140行，get_by_idconfluent.schemaregistry.client.ClientError：从注册表接收到错误的模式。

kafka connect oracle是否将未验证的模式发送到模式注册表？如何将此模式转换为正确的格式？

谢谢

Answer 1

看起来您的模式有问题。JSON格式化程序说它是无效格式。您可以在此处检查JSON格式是否正确：https://jsonformatter.curiousconcept.com/#

通过查看它，我看到这里有两个过度使用的引号：

第一个在第一行，在“schema”之后：

第二个在最后一行，在<代码>测试之间。行“}和，“deleted”：false}

删除这两个后，它现在是有效的形式。如果你问一个方法来自动这样做，我不知道怎么做。也许您可以搜索一些python代码来验证和修复JSON格式。

这是有效的格式：

{
   "subject":"TEST.KAFKAUSER.TEST-value",
   "version":1,
   "id":44,
   "schema":{
      "type":"record",
      "name":"row",
      "namespace":"test.kafkauser.test",
      "fields":[
         {
            "name":"SCN",
            "type":"long"
         },
         {
            "name":"SEG_OWNER",
            "type":"string"
         },
         {
            "name":"TABLE_NAME",
            "type":"string"
         },
         {
            "name":"TIMESTAMP",
            "type":{
               "type":"long",
               "connect.version":1,
               "connect.name":"org.apache.kafka.connect.data.Timestamp",
               "logicalType":"timestamp-millis"
            }
         },
         {
            "name":"SQL_REDO",
            "type":"string"
         },
         {
            "name":"OPERATION",
            "type":"string"
         },
         {
            "name":"data",
            "type":[
               "null",
               {
                  "type":"record",
                  "name":"value",
                  "namespace":"",
                  "fields":[
                     {
                        "name":"ID",
                        "type":[
                           "null",
                           "double"
                        ],
                        "default":null
                     },
                     {
                        "name":"NAME",
                        "type":[
                           "null",
                           "string"
                        ],
                        "default":null
                     }
                  ],
                  "connect.name":"value"
               }
            ],
            "default":null
         },
         {
            "name":"before",
            "type":[
               "null",
               "value"
            ],
            "default":null
         }
      ],
      "connect.name":"test.kafkauser.test.row"
   },
   "deleted":false
}