Introduction

When Flink writes records to Kafka, it needs a serialization layer that converts your domain object into the bytes Kafka will store. For Kafka 2.2, the usual answer with FlinkKafkaProducer is to implement either a simple SerializationSchema for value-only writes or a KafkaSerializationSchema when you need control over topic, key, partition, or headers.

Choose the Right Schema Interface

A plain SerializationSchema<T> is enough when every record goes to one topic and you only need a value payload.

1import org.apache.flink.api.common.serialization.SerializationSchema;
2
3public final class EventValueSerializer implements SerializationSchema<Event> {
4    @Override
5    public byte[] serialize(Event event) {
6        return (event.getId() + "," + event.getType()).getBytes(java.nio.charset.StandardCharsets.UTF_8);
7    }
8}

That works, but it gives you only the value bytes. In real Kafka pipelines you often need a key for partitioning or a dynamic topic choice. For that, use KafkaSerializationSchema<T>.

Implement KafkaSerializationSchema

KafkaSerializationSchema lets you return a full Kafka ProducerRecord. That is usually the better fit for Kafka 2.2 because it maps directly to the Kafka producer model.

1import java.nio.charset.StandardCharsets;
2import javax.annotation.Nullable;
3import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
4import org.apache.kafka.clients.producer.ProducerRecord;
5
6public final class EventKafkaSerializer implements KafkaSerializationSchema<Event> {
7    @Override
8    public ProducerRecord<byte[], byte[]> serialize(Event event, @Nullable Long timestamp) {
9        String topic = event.isPriority() ? "priority-events" : "events";
10        byte[] key = event.getId().getBytes(StandardCharsets.UTF_8);
11        byte[] value = (event.getType() + "," + event.getPayload()).getBytes(StandardCharsets.UTF_8);
12        return new ProducerRecord<>(topic, key, value);
13    }
14}

This approach is easier to evolve because the serializer owns the Kafka-specific details while the rest of the Flink job stays focused on data flow.

Wiring the Producer Into a Flink Job

After defining the schema, create the producer with Kafka properties and attach it to the stream.

1import java.util.Properties;
2import org.apache.flink.streaming.api.datastream.DataStream;
3import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
4
5Properties props = new Properties();
6props.setProperty("bootstrap.servers", "localhost:9092");
7props.setProperty("acks", "all");
8props.setProperty("retries", "3");
9
10FlinkKafkaProducer<Event> producer = new FlinkKafkaProducer<>(
11    "events",
12    new EventKafkaSerializer(),
13    props,
14    FlinkKafkaProducer.Semantic.AT_LEAST_ONCE
15);
16
17DataStream<Event> stream = env.fromElements(
18    new Event("1", "created", "alpha", false),
19    new Event("2", "updated", "beta", true)
20);
21
22stream.addSink(producer);

The topic string passed to the constructor is still required by some constructor variants, even if your serializer chooses the topic dynamically. Treat it as a default rather than as the only destination.

Delivery Semantics Matter More Than Serialization

Serialization gets most of the attention because it is the visible code, but producer semantics often matter more operationally. AT_LEAST_ONCE is simpler and sufficient for many pipelines. EXACTLY_ONCE requires checkpointing and broker support, and it increases coordination cost.

If you are debugging duplicates, do not assume the serializer is wrong. The issue may come from checkpoint configuration, retries, or downstream idempotency.

Keep the Payload Contract Stable

A serializer is part of a data contract. If you change field order, delimiters, or encodings casually, consumers break. For anything beyond quick demos, prefer an explicit format such as JSON, Avro, or Protobuf.

A JSON variant is still simple:

1String valueText = String.format(
2    "{\"id\":\"%s\",\"type\":\"%s\",\"payload\":\"%s\"}",
3    event.getId(),
4    event.getType(),
5    event.getPayload()
6);
7byte[] value = valueText.getBytes(StandardCharsets.UTF_8);

That is not as robust as a schema registry-backed format, but it is at least self-describing and easier to inspect.

Common Pitfalls

• Implementing SerializationSchema when you actually need to set Kafka keys or route to multiple topics. Use KafkaSerializationSchema in those cases.
• Forgetting that Kafka partitioning depends on the serialized key bytes. An unstable key format leads to unstable partition distribution.
• Treating serialization bugs and delivery-semantics bugs as the same problem. Duplicates are often caused elsewhere.
• Hard-coding a text format without documenting it. Producers and consumers need a stable payload contract.
• Ignoring character encoding. Use UTF-8 explicitly instead of relying on platform defaults.

Summary

• Use SerializationSchema only for simple value-only writes.
• Use KafkaSerializationSchema when you need topic, key, or record-level control.
• Build ProducerRecord objects directly for Kafka-oriented behavior.
• Configure delivery semantics separately from serialization logic.
• Treat the serialized payload as a versioned contract, not as incidental string concatenation.