由于架构不兼容，Kafka JDBC 源连接器无法注册架构

问题描述

我们有一个表，其中包含不可空列的 modifiedTime 列。我已经从中创建了一个视图，并将该列命名为 rowLastUpdated。此视图会同步到 Kafka。使用的数据库是Oracle 12c。

我在连接器状态中看到了这个问题：

Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException:
     Schema being registered is incompatible with an earlier schema; error code: 409

我查询了此视图的架构注册表，但只注册了一个架构。在缩小问题范围时，我只粘贴了 rowLastUpdated 的 avro 架构。

{
  "name":"ROWLASTUPDATED","type":[
    "null",{
      "type":"long","connect.version":1,"connect.name":"org.apache.kafka.connect.data.Timestamp","logicalType":"timestamp-millis"
    }
  ],"default":null
}

后续迭代中的 kafka 源连接器注册了一个新架构，其中 rowLastUpdated 如下所示（我从连接器任务状态的堆栈跟踪中获得了以下架构）：

    {
      "name":"ROWLASTUPDATED","type":{
         "type":"long","logicalType":"timestamp-millis"
      }
   }

我们可以观察到注册是在没有类型 null 的情况下发生的。这导致了架构不兼容问题。

原始表中的 modifiedTime 列从第 1 天起就是不可为空的列。我想知道 kafka connect 如何为同一个视图生成 2 个不同的模式。有人可以帮我找出问题所在吗？

这是源连接器配置

connector.class : io.confluent.connect.jdbc.JdbcSourceConnector
timestamp.column.name : ROWLASTUPDATED
incrementing.column.name : IDCOL
errors.retry.timeout : -1
validate.non.null : false
connection.password : ...
timestamp.initial : 0
tasks.max : 1
connection.attempts : 120
batch.max.rows : 2000
errors.retry.delay.max.ms : 60000
timestamp.delay.interval.ms : 600000
connection.backoff.ms : 30000
table.types : view
table.whitelist : ...
mode : timestamp+incrementing
topic.prefix : view-
connection.user : ...
schema.pattern : ...
poll.interval.ms : 300000
name : connector-Test
connection.url : jdbc:oracle:thin:@//x.x.x:1521/x.x.x
fetch.size : 5000

我已将 schema.pattern 设置为源数据库用户。目标数据库也有同名的表，但用户不同。我已经检查过了。因此，此源连接器已正确配置为仅查询源视图。

更新

我正在通过 kafka 连接代码来了解如何从 ResultSet 生成架构。这就是他们所做的 - https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/GenericDatabaseDialect.java#L729。所以，我认为问题可能出在 ojdbc8 驱动程序上。我提出了一个单独的问题 - Oracle JDBC driver - ResultSetMetadata isNullable gives incorrect results

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

apache-kafka apache-kafka-connect confluent-schema-registry ojdbc oracle12c