SAMZA-778: Fix KafkaCheckpointMigration code to avoid start/stop Coor…

…dinatorStreamSystemConsumer multiple times
cc13ny · Sep 28, 2015 · 2b12242 · 2b12242
1 parent c76132f
commit 2b12242
Showing 1 changed file with 0 additions and 1 deletion.
diff --git a/samza-kafka/src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala b/samza-kafka/src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala
@@ -79,7 +79,6 @@ class KafkaCheckpointMigration extends MigrationPlan with Logging {
     coordinatorSystemConsumer.start()
     coordinatorSystemConsumer.bootstrap()
     val stream = coordinatorSystemConsumer.getBootstrappedStream(SetMigrationMetaMessage.TYPE)
-    coordinatorSystemConsumer.stop()
     val message = new SetMigrationMetaMessage(source, migrationKey, migrationVal)
     stream.contains(message.asInstanceOf[CoordinatorStreamMessage])
   }