Problem

Discussed here

Right now, most of the sinks that do partitioning do not update the event status as EventStatus::Rejected and silently filter out events that fails to be partitioned.. Then (somehow, I don't know where this happens) the finalizer status is updated to EventStatus::Delivered

For example

for there are a lot more sinks that just do .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) }) to filter out None partition key values.

Partition error handling should be improved, I propose to change the signature of the Partitioner trait so the Partitioner::partition method returns Result<Self::Key> and that result should be handled (updating the finalizer's status to EventStatus::Rejected) and filtered out in PartitionedBatcher::poll which is what is called in SinkBuilderExt::batched_partitioner

When this issue is addressed, this TODO needs to be updated

The impact to users is that acknowledgements are not working as expected when partitioning fails (for example, rendering a template), as the event failed to process but it is reported that it went ok.

Version

0.49.0