Kafka Connect：主题显示的事件数量是预期的 3 倍答案

【问题标题】：Kafka Connect: Topic shows 3x the number of events than expectedKafka Connect：主题显示的事件数量是预期的 3 倍
【发布时间】：2020-04-08 17:57:15
【问题描述】：

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

同步通常工作正常，但似乎存储在主题中的事件/消息数量是预期的 3 倍。

这可能是什么原因？

一些附加信息

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

大部分主题分为3个分区（通过SMT设置Key，使用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"
  }
}

奇怪的话题数

【问题讨论】：

你为什么使用timestamp+incrementing模式为什么不只是timestamp？
正确 - timestamp+incrementing 是一个不错的选择
我可以为此推荐 KSQL :) 然后您可以针对数据运行 COUNT...GROUP BY。
您能否编辑您的问题以以纯文本形式包含您的源和接收器连接器配置？
是所有主题都有重复项还是只是其中一些？

标签： apache-kafka apache-kafka-connect

【解决方案1】：

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

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

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

在调查方面我建议：

通过将连接器拆分为每个表一个连接器来隔离问题。
检查每个主题并找到重复消息的示例。查看是否存在主题重复的模式。 KSQL 在这里会很有用：
```
SELECT ROWKEY, COUNT(*) FROM source GROUP BY ROWKEY HAVING COUNT(*) > 1
```
我猜ROWKEY（Kafka 消息的键）- 你会知道你的数据以及哪些列应该是唯一的并且可以用来检测重复。
找到重复消息后，使用 kafkacat 检查重复实例。它们是否具有完全相同的 Kafka 消息时间戳？

更多来回，StackOverflow 不是一个合适的平台 - 我建议前往 http://cnfl.io/slack 和 #connect 频道。

【讨论】：