Kafka连接：主题显示的事件数是预期的3倍

Question

我们正在使用Kafka Connect JDBC将表同步到数据库（Debezium非常适合，但不可能）。

Sync通常运行良好，但似乎存储在主题中的事件/消息数量是预期的3倍。

这可能是什么原因？

一些附加信息

目标数据库包含确切的消息数（主题中的消息数/3）。

大多数主题分为3个分区（键通过SMT设置，使用DefaultPartitioner）。

JDBC源连接器

{
  "name": "oracle_source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:oracle:thin:@dbdis01.allesklar.de:1521:stg_cdb",
    "connection.user": "****",
    "connection.password": "****",
    "schema.pattern": "BBUCH",
    "topic.prefix": "oracle_",
    "table.whitelist": "cdc_companies, cdc_partners, cdc_categories, cdc_additional_details, cdc_claiming_history, cdc_company_categories, cdc_company_custom_fields, cdc_premium_custom_field_types, cdc_premium_custom_fields, cdc_premiums, cdc, cdc_premium_redirects, intermediate_oz_data, intermediate_oz_mapping",
    "table.types": "VIEW",
    "mode": "timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "ts",
    "key.converter": "org.apache.kafka.connect.converters.IntegerConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "validate.non.null": false,
    "numeric.mapping": "best_fit",
    "db.timezone": "Europe/Berlin",
    "transforms":"createKey, extractId, dropTimestamp, deleteTransform",
    "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
    "transforms.createKey.fields": "id",
    "transforms.extractId.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractId.field": "id",
    "transforms.dropTimestamp.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.dropTimestamp.blacklist": "ts",
    "transforms.deleteTransform.type": "de.meinestadt.kafka.DeleteTransformation"
  }
}

JDBC接收器连接器

{
  "name": "postgres_sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:postgresql://writer.branchenbuch.psql.integration.meinestadt.de:5432/branchenbuch",
    "connection.user": "****",
    "connection.password": "****",
    "key.converter": "org.apache.kafka.connect.converters.IntegerConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.schemas.enable": true,
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "id",
    "delete.enabled": true,
    "auto.create": true,
    "auto.evolve": true,
    "topics.regex": "oracle_cdc_.*",
    "transforms": "dropPrefix",
    "transforms.dropPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.dropPrefix.regex": "oracle_cdc_(.*)",
    "transforms.dropPrefix.replacement": "$1"
  }
}

奇怪的主题计数

Answer 1

这本身不是一个答案，但在这里比在评论框中更容易格式化。

不清楚你为什么会得到重复的。一些可能性是：

1. 您有多个连接器实例正在运行
2. 您有正在运行的连接器实例，但之前已经运行了将相同数据加载到主题的其他实例
3. 数据来自多个表并被合并到一个主题中（根据您的配置，这里不可能，但如果您使用单消息转换来修改目标主题名称可能是可能的）

在调查方面，我建议：

>

SELECT ROWKEY, COUNT(*) FROM source GROUP BY ROWKEY HAVING COUNT(*) > 1

我在猜测ROWKEY（Kafka消息的键）-您将知道您的数据以及哪些列应该是唯一的，并且可以用于检测重复项。

要了解更多信息，StackOverflow并不是一个合适的平台-我建议前往http://cnfl.io/slack和#连接通道。