Hey everyone 🙂 I have a usecase where I have a single flink application who listens to kafka topics with some consumer group id. I want to duplicate that app to listen to the same topics, and each duplicate should read all of the messages in the topic. Therefore, I need to change the consumer group id for each one of the consumers. When doing the migration (stopping the single app and initializing the duplicate ones), I want each of them to start at the last committed offsets by the previous single app consumer id. What's the best approach here?

Therefore, I need to change the consumer group id for each one of the consumers

That’s not necessarily required. Flink keeps offsets in the state and it doesn’t rely on consumer groups for offset management. If you simply start a new job without state it can read all messages in the topic every time.

Actually when I remove the single app and start the duplicate ones, I need them to start from the offset that the single app stopped at

Then you can just take a savepoint, stop it, start a new app from the savepoint.

There are cases where you can't, such as change in the structure of the state

Got it, then you can use this workflow: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kafka/#upgrading-to-the-latest-connector-version