One bad Kafka record shouldn't crash a Flink Stateful Functions job

In Apache Stateful Functions, a single malformed Kafka record could crash the entire Flink job. Previously, the routable Kafka ingress had no policy for handling malformed records, causing a null key to throw an error within the deserializer. A tombstone, which is a normal compacted topic occurrence, would also result in a NullPointerException deep inside protobuf with no information about the specific record that caused the issue.

As a result, every ingress, topic, and function within the Flink job would fail. This loop continued as the poison record's offset was never committed, leading the job to terminate with a FAILED status. Apache Stateful Functions version 3.4.0-KZM-3.5 introduced the invalidRecordHandling feature in io.statefun.kafka.v1/ingress policies.

This new feature allows for either skipping the invalid record with a warning log or failing the job with a strict contract. The skipped record is logged with detailed coordinates, including topic, partition, offset, timestamp, key, and value size. This information is useful for alerting systems to identify misbehaving producers and the type of corruption.

Two policies are available: a default policy applied to all topics, and a per-topic override policy. The default log level is set to warn, but it can be adjusted to debug, info, or error. The alerting system focuses on the numInvalidRecordsSkipped counter instead of the job restarts. Custom KafkaIngressDeserializer implementations that return a null value will now also skip the record instead of crashing the job.

Future improvements include delivering invalid records to a dead-letter function with provenance metadata, allowing pipelines to quarantine or replay them.