Kafka tombstone message support

[vlad-lyutenko]

Description

Kafka tombstone message support

Additional context and related issues

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Kafka
* Fix query failure when a Kafka topic contains tombstone messages, that is messages with a ``NULL`` value. ({issue}`16962`)

[hashhar]

LGTM

[hashhar]

do you think a test which verifies that tombstones can be differentiated from messages with all fields set to null would be useful? It would help us "document" the contract which we talked about so that users can differentiate between tombstones and actual nulls in their messages.

[vlad-lyutenko]

Test added, we can differentiate these 2 cases by internal message_corrupt field

[hashhar]

@mosabua this would be nice to document on how users can inspect/filter out tombstone messages if they want to. Before this change it lead to query failure.

The TL;DR is:

• A tombstone message is a message in Kafka which has a null value with a non-null key k, often emitted to indicate that the record k was deleted
• When querying a Kafka topic which contains tombstone messages and messages which have all their fields set to null you can differentiate between them by inspecting the _message_corrupt field. It's set to true for tombstone messages. So you can find out tombstone messages by doing something like WHERE (field1 IS NULL AND field2 IS NULL .. AND field_n IS NULL AND _message_corrupt = true). Just _message_corrupt = true will also match actually corrupt messages so by itself it's not a sufficient filter.