[Bug Report]:Unable to produce tombstone records about kafkaflow HOT 7 OPEN

When fixing it with the serializer, then it has to be changed for all serializers. Sending null for tombstoning should not be part of the serialization imho.

great library by the way. :)

from kafkaflow.

Do you have any error while producing the message?

from kafkaflow.

No, there is no error. But it won't produce tombstone records when using a kafka upsert source: https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/upsert-kafka/ .

from kafkaflow.

Like we discussed on kafkaflow on slack, by using a custom serializer, you can bypass this limitation by checking if the message is null before serializing the message value.

This should produce a message where the message value is an empty byte[] which is the tombstone.

Meanwhile, we will address this issue, by creating a pull request to fix the JsonCoreSerializer.

Regarding the Kakfa upsert source, flink we won't expect a different behavior from Kafka client as they both produce a null record.

from kafkaflow.

I have tried that approach, end to end, using kafka-upsert source and iceberg table destination in AWS. It sends indeed an empty byte array, but that did not translate into a tombstone record.
If I read correctly this https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/upsert-kafka/, it would only work if we send null for message.
Maybe I missed something, but I'm happy to test your PR once ready.

from kafkaflow.

Fixing the issue with JsonCoreSerializer when producing a message with a null value it results in a tombstone record where the message value is an empty byte[].

"A null payload is a payload with 0 bytes"
reference here

public Task SerializeAsync(object? message, Stream output, ISerializerContext context)
    {
        if (message == null)
        {
            return Task.CompletedTask;
        }

        return SerializeNonNullMessageAsync(message, output);
    }

As an alternative, you can always use the native Kafka client to confirm that it also produces a tombstone record with an empty byte[] like KafkaFlow.

from kafkaflow.