Simply like this: env.addSource(new FlinkKafkaConsumer(...)).uid(“some-unique-id”)
The same goes for any other operator. However, do keep in mind this bug that was just recently uncovered: https://issues.apache.org/jira/browse/FLINK-7623. What I described in my previous reply would not work as expected if your Kafka consumer was chained to some stateful operator. The workaround, for now, would be to explicitly disable chaining of the consumer source with any stateful operators before taking the savepoint and changing the operator UID. If your consumer source is not chained with any stateful operator, then you can ignore this and safely proceed. Cheers, Gordon On 14 September 2017 at 3:49:31 PM, kla (lalafar...@gmail.com) wrote: Hi Gordon, Thanks for your quick reply. I have following consumer: jobConfiguration.getEnv().addSource( new FlinkKafkaConsumer010<>(properties.getProperty(TOPIC), deserializer, properties)); How can I set the UID for the consumer ? Thanks again for help! Regards, Konstantin -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
