使用kafka的自动删除行？

Question

我有一个内存数据库，我正在使用KAFKA + JDBCSINKCONNECTOR将下游Postgres数据库与内存数据库同步。内存数据库用于有效的计算，而Postgres则用于查询。在开发中，我经常破坏并重新创建内存数据库。每次，我还会重新创建Kafka水槽连接器。

如果添加了新行或在内存数据库中更改了现有行，我认为JDBCSinkConnector能够将Postgres与新数据同步。但是，如果删除行，JDBCSinkConnector不会删除Postgres中的行。

JDBCSinkConnector是否可以检查下游数据库中的哪个行不再在上游数据库中，然后删除它们？如果没有，每次更新上游数据库时，我都必须销毁下游数据库。

配置：

{
  'connector.class': 'io.confluent.connect.jdbc.JdbcSinkConnector',
  'dialect.name': 'PostgreSqlDatabaseDialect',
  'key.converter': 'io.confluent.connect.avro.AvroConverter',
  'key.converter.schema.registry.url': `http://schema-registry:${process.env.SCHEMA_REGISTRY_PORT}`,
  'value.converter': 'io.confluent.connect.avro.AvroConverter',
  'value.converter.schema.registry.url': `http://schema-registry:${process.env.SCHEMA_REGISTRY_PORT}`,
  'insert.mode': 'upsert',
  'delete.enabled': 'true',
  'auto.create': 'true',
  'auto.evolve': 'false',
  'errors.retry.timeout': -1,
  'connection.url': `jdbc:postgresql://${process.env.INTERNAL_DOCKER_HOST}:${process.env.PG_PORT}/${process.env.PG_DB}`,
  'connection.user': process.env.PG_USER,
  'connection.password': process.env.PG_PASS,
  'pk.mode': 'record_key',
}

Answer 1

JDBCSinkConnector可能不再在上游数据库中检查下游数据库中的哪个行

与上游系统的任何知识完全脱钩，所以没有。

每次更新上游数据库时，我都必须销毁下游数据库。

截断，不要销毁/掉落。除非您的上游数据库架构更改。您需要一些外部通知事件来触发这一点；可能是REST API，不一定是Kafka事件

---

不清楚，为什么当Kafka流ktables还允许主题快速存储和支持时，为什么需要使用物质化，并且它们支持写入JDBC接收器所需的墓碑事件来删除数据。