Doing this here could lead to potentially issues in the non-tombstone cases, don't you think?

I can see your concern this is generic code that is used in other places, though if this is a non-tombstone event this if condition will be skipped and the original path will be executed. If this is a tombstone then the there is no value struct as that's part of the isTombstone test so we would without this code return null.
I also checked and there's no after structure when we are dealing with a key. I think this is safe from my testing

this is no longer required

I would put this out avoiding the return value since, as the method name suggests, it will process the delete.

@jpechane @Naros I am currently hitting the question why deletes are disabled by default?

@rk3rn3r, I aligned this with Confluent's implementation to minimize the configuration changes for those who move from one to the other.

To me this also applies to other configurations like schema.evolution being disabled by default and the insert.mode defaulting to insert rather than upsert. Again, mirroring the same config defaults from Confluent.

@msillence Should we maybe move the below code for if (record.isDelete()) { ... } up here? record.isDelete() covers the case of tombstones and Debezium delete change event records. The code below can be deleted.

Also @jpechane @mfvitale @Naros Please mind that this new behavior will cause 2 deletes executed because tombstones.on.delete in source connectors and even with the deprecated delete.handling​.mode option in ExtractNewRecordState SMT COULD cause 2 messages for a single delete event happening in the source datastore (one Debezium DELETE change event and one tombstone), depending on those configs. The implementation I am working on will de-duplicate those events so this is only temporary an issue. The reason for the previous code was that we avoided duplicate delete events for the same row/source event. But it excludes deletes for source connectors using ExtractNewRecordState SMT that drops the Debezium DELETE events and keeps tombstones (which might be semantically the best way to represent deletes in Kafka imho).

@jpechane As this code is moved to the AbstractChangeEventSink eventually with my current work, we should maybe evaluate if we close this PR as those lines will be dropped soon?
Also we should have tests then for those scenarios with this PR (single tombstone, single Debezium DELETE change event, and DELETE event followed by tombstone). FYI @msillence

@msillence After writing the above, maybe we can just drop those lines here for if (record.isTombstone()) {...} entirely and keep the if (record.isDelete()) { ... } below as it handles tombstones and Debezium DELETE change event messages, yielding the previously mentioned duplicate deletes issued to the destination datastore. The idea was to skip those duplicates delete events in the default scenarion but excludes for certain ExtractNewRecordState SMT configs (which imho are even syntactically and semantically good and correct)!

@jpechane I am sorry for the late coming to the party.
@rk3rn3r I'd like to get this into 3.0 as it is the next downstream version. Looking a the calendar I'd suppose your changes will go into 3.1.

Generally the new solution must

• Exactly one DELETE statement should be issued if the SMT is not in place
• Exactly one DELETE statement should be issued if SMT is in place and delete.tombstone.handling.mode = drop
• Exactly one DELETE statement should be issued if SMT is in place and delete.tombstone.handling.mode = rewrite-with-tombstone

All three cases should be tested. If this would require a change at either core or SMT then it should be done. The legacy config options could be ignored and thei behaviour in this case exteced as undefined.