How can I add message key to KafkaSink in Apache Flink 1.14

user_15958302 0
Viewed 301

As stated in the title I need to set a custom message key in KafkaSink. I cannot find any indication on how to achieve this in the Apache Flink 1.14 docs. At the moment I'm correctly setting up the KafkaSink and the data payload is correctly written in the topic, but the key is null. Any suggestions? Thanks in advance

flink-streaming apache-flink
1 Answers

You should implement a KafkaRecordSerializationSchema that sets the key on the ProducerRecord returned by its serialize method.

You'll create the sink more-or-less like this:

KafkaSink<UsageRecord> sink =
    KafkaSink.<UsageRecord>builder()
             .setBootstrapServers(brokers)
             .setKafkaProducerConfig(kafkaProps)
             .setRecordSerializer(new MyRecordSerializationSchema(topic))
             .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
             .setTransactionalIdPrefix("my-record-producer")
             .build();

and the serializer will be something like this:

public class MyRecordSerializationSchema implements
        KafkaRecordSerializationSchema<T> {

    private static final long serialVersionUID = 1L;

    private String topic;
    private static final ObjectMapper objectMapper =
        JsonMapper.builder()
            .build()
            .registerModule(new JavaTimeModule())
            .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);

    public MyRecordSerializationSchema() {}

    public MyRecordSerializationSchema(String topic) {
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(
            T element, KafkaSinkContext context, Long timestamp) {

        try {
            return new ProducerRecord<>(
                    topic,
                    null, // choosing not to specify the partition
                    element.ts.toEpochMilli(),
                    element.getKey(),
                    objectMapper.writeValueAsBytes(element));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException(
                "Could not serialize record: " + element, e);
        }
    }
}

Note that this example is also setting the timestamp.

FWIW, this example comes from https://github.com/alpinegizmo/flink-mobile-data-usage/blob/main/src/main/java/com/ververica/flink/example/datausage/records/UsageRecordSerializationSchema.java.

user_20008230
Related