For datasource, I listen to multiple topics, when I receive a source event, how can I know which topic this event consumed from, since topic name contains some information I need (e.g. country prefix)
I was trying to read the source code of flink KafkaConsumer but didn't find any method I can use to get topic name from an event.
If you're listening to different topics with different names and you need that information, you should probably enrich your events with the information you need with a simple map.
For instance if your topic contains events with fields like "username" and "timestamp"
{
"username": "diego",
"timestamp": 1649085282804
}
You can listen to your source and make a map right away after each source to add the information you need to each event.
val enrichedDataStream: Datastream[CountryCodeEnrichedEvent] = kafkaSourceCanada.map(event => CountryCodeEnrichedEvent(username: event.username, timestamp: event.timestamp, countryCode: "ca"))
You'll end up with something like this, for each kafka source you're listening to.
{
"username": "diego",
"timestamp": 1649085282804,
"countryCode": "ca"
}
The topic is part of the ConsumerRecord that is passed to the deserialize method of a KafkaDeserializationSchema. So if you want to access this or other Kafka metadata, you'll need to implement your own deserializer.