Hey all,

I’m building a Flink app that pulls in messages from a Kafka topic and writes 
them out to disk using a custom bucketed sink. Each message needs to be parsed 
using a schema that is also needed when writing in the sink. This schema is 
read from a remote file on a distributed file system (it could also be fetched 
from a service). The schema will be updated very infrequently.

In order to support schema evolution, I have created a custom source that 
occasionally polls for updates and if it finds one parses the new schema and 
sends a message containing the serialized schema. I’ve connected these two 
streams and then use a RichCoFlatMapFunction to flatten them back into a single 
output stream (schema events get used to update the parser, messages get parsed 
using the parser and emitted).

However, I need some way to communicate the updated schema to every task of the 
sink. Simply emitting a control message that is ignored when writing to disk 
means that only one sink partition will receive the message and thus update the 
schema. I thought about sending the control message as side output and then 
broadcasting the resulting stream to the sink alongside the processed event 
input but I couldn’t figure out a way to do so. For now, I’m bundling the 
schema used to parse each event with the event, storing the schema in the sink, 
and then checking every event’s schema against the stored schema but this is 
fairly inefficient. Also, I’d like to eventually increase the types of control 
messages I can send to the sink, some of which may not be idempotent. Is there 
a better way to handle this pattern?


(Bonus question: ideally, I’d like to be able to perform an action when all 
sink partitions have picked up the new schema. I’m not aware of any way to emit 
metadata of this sort from Flink tasks beyond abusing the metrics system. This 
approach still leaves open the possibility of tasks picking up the new schema 
and then crashing for unrelated reasons thus inflating the count of tasks using 
a specific schema and moreover requires tracking at least the current level of 
parallelism and probably also Flink task state outside of Flink. Are there any 
patterns for reporting metadata like this to the job manager?)

I’m using Flink 1.8.

Reply via email to