Apache Flink: How to sink events to different Kafka topics depending on the event type?

Question

I was wondering if it was possible to use the Flink Kafka sink to write events different topic depending on the type of events? Let's say that we have different type of events: notification, messages and friend requests. We want to stream these events to different topics named: notification-topic, messages-topic, friendsRequest-topic.

I tried many different ways to resolve this problem, but still couldn't find the right solution. I heard that I could use the ProcessFunction but how can it be related to my problem?

Answer 1

In case you are using Kafka:

FlinkKafkaProducer011<Event> producer = new FlinkKafkaProducer011<>(
    "default.topic",

    new KeyedSerializationSchema<Event>() {
    @Override
    public byte[] serializeKey( Event element ) {
        return null; or element.getKey to bytes...
    }

    @Override
    public byte[] serializeValue( Event element ) {
            return event.toBytes() ...
    }

    @Override
    public String getTargetTopic( Event element ) {
        return element.getTopic();
    }
    },
    parameterTool.getProperties());

input.addSink(producer);

It will call getTargetTopic for every event, to get the topic where you want to route the event. It will override the "default.topic"