You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2015/09/28 19:18:40 UTC
samza git commit: SAMZA-778: Fix KafkaCheckpointMigration code to
avoid start/stop CoordinatorStreamSystemConsumer multiple times
Repository: samza
Updated Branches:
refs/heads/master c76132ff2 -> 2b12242c7
SAMZA-778: Fix KafkaCheckpointMigration code to avoid start/stop CoordinatorStreamSystemConsumer multiple times
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/2b12242c
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/2b12242c
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/2b12242c
Branch: refs/heads/master
Commit: 2b12242c7cc8d120ff33f0d99c94e56274ede9f8
Parents: c76132f
Author: Yi Pan <ni...@gmail.com>
Authored: Mon Sep 28 10:17:51 2015 -0700
Committer: Yi Pan (Data Infrastructure) <yi...@linkedin.com>
Committed: Mon Sep 28 10:18:22 2015 -0700
----------------------------------------------------------------------
.../src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/2b12242c/samza-kafka/src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala b/samza-kafka/src/main/scala/old/checkpoint/KafkaCheckpointMigration.scala
index 2e10815..32afe4c 100644
--- 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])
}