feat: Allow empty payloads in Kafka, in order to support compaction tombstones

[mhelleborg]

In order to support deletes for compacted topics, this PR allows empty / null payloads to be sent when using the Kafka extension. The previous requirement that the payload is present has been dropped, and tests added.

Ref Log compaction

[jskeet]

I'll have a look at this when I get the chance, but I'm afraid it may be a while due to vacations. (I don't know anything about Kafka, which means it'll take me longer to try to work out whether I think it's a good idea or not...)

[mhelleborg]

Thanks Jon :)

The change should make it easy to use CloudEvents for projecting state using compacted topics, as a streaming KV-store.

The zero-length payloads are treated as tombstones, and lets Kafka clean up removed messages over time.

[jskeet]

I still don't know why we prohibited empty messages in the first place. I'm going to ask at the meeting tomorrow. If other SDKs are fine with empty payloads, we should be too. If they're not, I want to find out why :)

[pierDipi]

Commenting after the WG discussion.

I agree with this PR.

In general, Kafka records with a null value are valid records with both log compacted topics or with regular topics.

However, in the log compacted topics case, a record with a null value has a different semantic (tombstone), they signal to Kafka that it can remove all previous records with the same key but it retains and delivers to subscribers the tombstone record as long as it is consumed before the the topic-level configuration delete.retention.ms expires.

Since configuring a topic to be log compacted is an event producer's responsibility, the producer is specifically asking for that behavior so I don't see why it shouldn't be allowed to send a CloudEvent that is a tombstone given that:

• the event itself is just a normal CloudEvent
• the event producer is specifically asking the transport to have that side effect

[jskeet]

Just to set expectations, I'm hoping to merge this PR next week - I'm on vacation tomorrow, and done with work for today.

[jskeet]

Approved in general, with a few nits.

I'm going to create a PR for the Kafka binding spec at https://github.com/cloudevents/spec/blob/main/cloudevents/bindings/kafka-protocol-binding.md to mention that an empty payload can have a special meaning when using binary mode.

[jskeet]

While not strictly invalid as far as I'm aware, is there any good reason to have a data content type when there isn't any data?

[jskeet]

(It looks like you're using that to effectively check the result of decoding different empty bodies - I don't think that's really part of the Kafka code, so I think you can probably simplify this test. If you'd prefer to keep it, that's okay.)

[mhelleborg]

I was thinking it's a good idea to verify that the deserialization is well behaved across all the content types, even without a payload.

I can add the null content type as well, to be sure that it works correctly too.

[jskeet]

But isn't that deserialization purely in the formatter, not in the Kafka code at all?

[mhelleborg]

True, it might be a case of over-testing. But I saw it as a better safe than sorry to cover it.

I can remove the content type checks if you wish?

[jskeet]

I don't mind too much - it just seems unnecessary. I haven't looked at what the coverage is like at the moment.