Details
-
Bug
-
Status: Resolved
-
Major
-
Resolution: Fixed
-
None
-
None
-
None
Description
The current implementation of KafkaSystemConsumer only allows start()/stop() once in the life cycle of the same SystemConsumer instance. Unfortunately, KafkaCheckpointMigration code start()/stop() the same instance of KafkaSystemConsumer multiple times, which results in an infinite loop in CoordinatorStreamSystemConsumer.start() in JobRunner.
The quick fix is to avoid the sequence of SystemConsumer.start()>SystemConsumer.stop()>SystemConsumer.start() in the KafkaCheckpointMigration code.
In long-term, we may also want to evaluate the use cases where we support start/stop multiple times in the life cycle of a single instance of SystemConsumer.