我正在尝试使用kafka connect实时复制一个表。使用的数据库是mysqlv5.7。分别使用插入和更新模式时，列的行为与预期一致。但是，当我使用upsert模式时，在数据库中没有观察到任何变化。
通过用户界面填写配置文件
下沉

topic = custom-p2p
Connector Class = JdbcSinkConnector  
name = sink
tasks-max = 1
Key-converter-class=org.apache.kafka.connect.storage.StringConverter
Value-converter-class=org.apache.kafka.connect.json.JsonConverter
jdbc_url=jdbc:mysql://127.0.0.1:3306/p2p_service_db4?user=root&password=root&useSSL=false
insert mode = upsert
auto create = true
auto evolve = true

来源

Connector Class = JdbcSourceConnector
name = source-new
task max = 1
key converter class = org.apache.kafka.connect.storage.StringConverter
value converter class = org.apache.kafka.connect.json.JsonConverter
jdbc url = jdbc:mysql://127.0.0.1:3306/p2p_service_db3?user=root&password=root&useSSL=false
table loading mode = timestamp+incrementing
incrementing column name = auto_id
timestamp column name = last_updated_at
topic prefix = custom-

我遇到的问题是，当sink insert模式更改为insert时，当更改为update时插入会正确进行，这也会像预期的那样完美地发生，但是当值更改为upsert时，插入和更新都不会发生。如果做错了什么，请告诉我？为什么此模式不起作用？如果这两个插入和更新都需要在备份数据库中复制，是否有其他替代方法。
先谢谢你。如果需要其他信息，请告诉我