Kafka-connect上的硬删除事件同步数据库不起作用或出现错误。

实现正确的'消息键'规则，其中如果同一事件在连续的时间内获得相同的消息键，则该事件将被视为一次性消除事件，并被正确从目标数据库中删除。

示例代码如下：

public class SinkTask extends JdbcSinkTask {

    private static final String MESSAGE_KEY_FIELD = "message.key.field.name";
    private String messageKeyFieldName;

    @Override
    public void start(Map<String, String> props) {
        super.start(props);
        messageKeyFieldName = props.get(MESSAGE_KEY_FIELD);
    }

    @Override
    public void put(Collection<SinkRecord> records) {
        List<SinkRecord> singularRecords = new ArrayList<>();

        for (SinkRecord record : records) {
            Object messageKey = record.valueSchema().field(messageKeyFieldName).schema().defaultValue();

            if (messageKey == null) {
                singularRecords.add(record);
                continue;
            }

            if (getPreviousMessageKey(messageKey) != null) {
                log.debug("Hard delete event received with an existing message key, skipping: {}", record);
                continue;
            }

            singularRecords.add(record);
            addPreviousMessageKey(messageKey);
        }

        super.put(singularRecords);
    }

    private Set<Object> previousMessageKeys = new HashSet<>();

    private Object getPreviousMessageKey(Object messageKey) {
        for (Object previousMessageKey : previousMessageKeys) {
            if (previousMessageKey.equals(messageKey)) {
                return previousMessageKey;
            }
        }
        return null;
    }

    private void addPreviousMessageKey(Object messageKey) {
        previousMessageKeys.add(messageKey);
    }
}