如何在JDBC Source Connector中将偏移量设置为开始，以便在Kafka中从MySQL表中获取所有数据

Question

我使用Kafka Connect添加了JDBC Source Connector，Kafka Connect从MySQL表中获取数据。

连接器添加成功，数据在其中实时流动。

但相应的主题并不包含所有先前的数据。

我尝试过重启Kafka Connect。

下面是我的JDBC源连接器配置：

{
"name": "kb_yp_loan",
"connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector",
"tasks.max": "1",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "XXXX",
"value.converter.basic.auth.credentials.source": "USER_INFO",
"value.converter.basic.auth.user.info": "XXXX",
"config.action.reload": "restart",
"connection.url": "jdbc:mysql://XXXX/XX?user=XXXX&password=XXXX&useCursorFetch=true&defaultFetchSize=1000",
"connection.user": "XXXX",
"connection.password": "XXXX",
"table.whitelist": "yp_loan",
"mode": "incrementing",
"incrementing.column.name": "id",
"topic.prefix": "kb_"

}

请建议我如何在我的Kafka主题中带来所有以前的数据。

Answer 1

在connect-distributed.properties中，您需要添加consumer.auto.offset.reset=earliest来启动连接消费者实例，以读取所有现有的主题数据。

否则，连接器将从当前的最新主题偏移量开始。