You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by sh...@apache.org on 2019/01/22 20:53:20 UTC
samza git commit: SAMZA-2074: Use configuration stored in coordinator
stream for CheckpointTool.
Repository: samza
Updated Branches:
refs/heads/master 759b7786a -> 5ce320c26
SAMZA-2074: Use configuration stored in coordinator stream for CheckpointTool.
Currently, each run of a `CheckpointTool` requires the `TaskModel` from the `JobModel` for updating the checkpoints. `JobModel` generation involves reading the `SystemStreamMetadata`, `SystemStreamPartitionMetadata` of the input streams of the job. Post samza 1.0, this would require the entire configuration bag stored in coordinator stream.
As a followup to `SAMZA-2059`, this PR changes the `CheckpointTool` to read the configuration from the coordinator stream and use it when generating the `JobModel`.
Author: Shanthoosh Venkataraman <sp...@usc.edu>
Reviewers: Cameron L<ca...@linkedin.com>
Closes #886 from shanthoosh/read_config_from_coordinator_stream_for_checkpoint_tool and squashes the following commits:
c0a68ec9 [Shanthoosh Venkataraman] Address review comments.
910cfaca [Shanthoosh Venkataraman] Read configuration from coordinator stream in checkpoint tool.
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/5ce320c2
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/5ce320c2
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/5ce320c2
Branch: refs/heads/master
Commit: 5ce320c26c0bebbccfcbaa6ed9e23b29d5ecd002
Parents: 759b778
Author: Shanthoosh Venkataraman <sp...@usc.edu>
Authored: Tue Jan 22 12:53:04 2019 -0800
Committer: Shanthoosh Venkataraman <sp...@usc.edu>
Committed: Tue Jan 22 12:53:04 2019 -0800
----------------------------------------------------------------------
.../samza/checkpoint/CheckpointTool.scala | 127 ++++++++++---------
.../samza/checkpoint/TestCheckpointTool.scala | 76 ++++++++---
2 files changed, 123 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/5ce320c2/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
index 65fb419..bb3b04d 100644
--- a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
+++ b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
@@ -22,16 +22,23 @@ package org.apache.samza.checkpoint
import java.net.URI
import java.util.regex.Pattern
+import joptsimple.ArgumentAcceptingOptionSpec
import joptsimple.OptionSet
import org.apache.samza.checkpoint.CheckpointTool.TaskNameToCheckpointMap
import org.apache.samza.config.TaskConfig.Config2Task
-import org.apache.samza.config.{Config, ConfigRewriter, JobConfig}
+import org.apache.samza.config.Config
+import org.apache.samza.config.ConfigRewriter
+import org.apache.samza.config.JobConfig
+import org.apache.samza.config.MapConfig
import org.apache.samza.container.TaskName
import org.apache.samza.job.JobRunner._
import org.apache.samza.metrics.MetricsRegistryMap
import org.apache.samza.system.SystemStreamPartition
-import org.apache.samza.util.{CommandLine, Logging, Util}
-import org.apache.samza.{Partition, SamzaException}
+import org.apache.samza.util.CommandLine
+import org.apache.samza.util.Logging
+import org.apache.samza.util.Util
+import org.apache.samza.Partition
+import org.apache.samza.SamzaException
import scala.collection.JavaConverters._
import org.apache.samza.coordinator.JobModelManager
@@ -70,12 +77,12 @@ import scala.collection.mutable.ListBuffer
object CheckpointTool {
/** Format in which SystemStreamPartition is represented in a properties file */
val SSP_PATTERN = "tasknames.%s.systems.%s.streams.%s.partitions.%d"
- val SSP_REGEX = Pattern.compile("tasknames\\.(.+)\\.systems\\.(.+)\\.streams\\.(.+)\\.partitions\\.([0-9]+)")
+ val SSP_REGEX: Pattern = Pattern.compile("tasknames\\.(.+)\\.systems\\.(.+)\\.streams\\.(.+)\\.partitions\\.([0-9]+)")
type TaskNameToCheckpointMap = Map[TaskName, Map[SystemStreamPartition, String]]
class CheckpointToolCommandLine extends CommandLine with Logging {
- val newOffsetsOpt =
+ val newOffsetsOpt: ArgumentAcceptingOptionSpec[URI] =
parser.accepts("new-offsets", "URI of file (e.g. file:///some/local/path.properties) " +
"containing offsets to write to the job's checkpoint topic. " +
"If not given, this tool prints out the current offsets.")
@@ -83,22 +90,22 @@ object CheckpointTool {
.ofType(classOf[URI])
.describedAs("path")
- var newOffsets: TaskNameToCheckpointMap = null
+ var newOffsets: TaskNameToCheckpointMap = _
def parseOffsets(propertiesFile: Config): TaskNameToCheckpointMap = {
var checkpoints : ListBuffer[(TaskName, Map[SystemStreamPartition, String])] = ListBuffer()
- propertiesFile.asScala.foreach { case (key, value) => {
+ propertiesFile.asScala.foreach { case (key, value) =>
val matcher = SSP_REGEX.matcher(key)
if (matcher.matches) {
val taskname = new TaskName(matcher.group(1))
val partition = new Partition(Integer.parseInt(matcher.group(4)))
val ssp = new SystemStreamPartition(matcher.group(2), matcher.group(3), partition)
- val tuple = (taskname -> Map(ssp -> value))
+ val tuple = taskname -> Map(ssp -> value)
checkpoints += tuple
} else {
warn("Warning: ignoring unrecognised property: %s = %s" format (key, value))
}
- }}
+ }
val taskNameSSPPairs = checkpoints.toList
if(taskNameSSPPairs.isEmpty) {
return null
@@ -111,7 +118,7 @@ object CheckpointTool {
.mapValues(m => m.reduce( _ ++ _)) // Merge all the maps of SSPs->Offset into one for the whole taskname
}
- override def loadConfig(options: OptionSet) = {
+ override def loadConfig(options: OptionSet): MapConfig = {
val config = super.loadConfig(options)
if (options.has(newOffsetsOpt)) {
val properties = configFactory.getConfig(options.valueOf(newOffsetsOpt))
@@ -121,29 +128,23 @@ object CheckpointTool {
}
}
- def apply(config: Config, offsets: TaskNameToCheckpointMap) = {
- val manager = config.getCheckpointManagerFactory match {
- case Some(className) =>
- Util.getObj(className, classOf[CheckpointManagerFactory])
- .getCheckpointManager(config, new MetricsRegistryMap)
- case _ =>
- throw new SamzaException("This job does not use checkpointing (task.checkpoint.factory is not set).")
- }
- new CheckpointTool(config, offsets, manager)
+ def apply(config: Config, offsets: TaskNameToCheckpointMap): CheckpointTool = {
+ val coordinatorStreamManager = new CoordinatorStreamManager(config, new MetricsRegistryMap())
+ new CheckpointTool(offsets, coordinatorStreamManager)
}
def rewriteConfig(config: JobConfig): Config = {
def rewrite(c: JobConfig, rewriterName: String): Config = {
val rewriterClassName = config
.getConfigRewriterClass(rewriterName)
- .getOrElse(throw new SamzaException("Unable to find class config for config rewriter %s." format rewriterName))
+ .getOrElse(throw new SamzaException("Config rewriter class: %s not found." format rewriterName))
val rewriter = Util.getObj(rewriterClassName, classOf[ConfigRewriter])
info("Re-writing config for CheckpointTool with " + rewriter)
rewriter.rewrite(rewriterName, c)
}
config.getConfigRewriters match {
- case Some(rewriters) => rewriters.split(",").foldLeft(config)(rewrite(_, _))
+ case Some(rewriters) => rewriters.split(",").foldLeft(config)(rewrite)
case _ => config
}
}
@@ -152,25 +153,34 @@ object CheckpointTool {
val cmdline = new CheckpointToolCommandLine
val options = cmdline.parser.parse(args: _*)
val config = cmdline.loadConfig(options)
- val rconfig = rewriteConfig(new JobConfig(config))
- print("Rewritten config" + rconfig)
- val tool = CheckpointTool(rconfig, cmdline.newOffsets)
- tool.run
+ val rewrittenConfig = rewriteConfig(new JobConfig(config))
+ info(s"Using the rewritten config: $rewrittenConfig")
+ val tool = CheckpointTool(rewrittenConfig, cmdline.newOffsets)
+ tool.run()
}
}
-class CheckpointTool(config: Config, newOffsets: TaskNameToCheckpointMap, manager: CheckpointManager) extends Logging {
+class CheckpointTool(newOffsets: TaskNameToCheckpointMap, coordinatorStreamManager: CoordinatorStreamManager) extends Logging {
+
+ def run() {
+ // Read the configuration stored in the coordinator stream.
+ coordinatorStreamManager.register(getClass.getSimpleName)
+ coordinatorStreamManager.start()
+ coordinatorStreamManager.bootstrap()
+ val configFromCoordinatorStream: Config = coordinatorStreamManager.getConfig
- def run {
- info("Using %s" format manager)
+ // Instantiate the checkpoint manager with coordinator stream configuration.
+ val checkpointManager: CheckpointManager = configFromCoordinatorStream.getCheckpointManagerFactory() match {
+ case Some(className) =>
+ Util.getObj(className, classOf[CheckpointManagerFactory])
+ .getCheckpointManager(configFromCoordinatorStream, new MetricsRegistryMap)
+ case _ =>
+ throw new SamzaException("Configuration: task.checkpoint.factory is not defined.")
+ }
// Find all the TaskNames that would be generated for this job config
- val coordinatorStreamManager = new CoordinatorStreamManager(config, new MetricsRegistryMap())
- coordinatorStreamManager.register(getClass.getSimpleName)
- coordinatorStreamManager.start
- coordinatorStreamManager.bootstrap
val changelogManager = new ChangelogStreamManager(coordinatorStreamManager)
- val jobModelManager = JobModelManager(coordinatorStreamManager.getConfig, changelogManager.readPartitionMapping())
+ val jobModelManager = JobModelManager(configFromCoordinatorStream, changelogManager.readPartitionMapping())
val taskNames = jobModelManager
.jobModel
.getContainers
@@ -179,46 +189,37 @@ class CheckpointTool(config: Config, newOffsets: TaskNameToCheckpointMap, manage
.flatMap(_.getTasks.asScala.keys)
.toSet
- taskNames.foreach(manager.register)
- manager.start
+ taskNames.foreach(checkpointManager.register)
+ checkpointManager.start()
- val lastCheckpoints = taskNames.map(tn => tn -> readLastCheckpoint(tn)).toMap
+ val lastCheckpoints = taskNames.map(taskName => {
+ taskName -> Option(checkpointManager.readLastCheckpoint(taskName))
+ .getOrElse(new Checkpoint(new java.util.HashMap[SystemStreamPartition, String]()))
+ .getOffsets
+ .asScala
+ .toMap
+ }).toMap
- lastCheckpoints.foreach(lcp => logCheckpoint(lcp._1, lcp._2, "Current checkpoint for taskname "+ lcp._1))
+ lastCheckpoints.foreach(lcp => logCheckpoint(lcp._1, lcp._2, "Current checkpoint for task: "+ lcp._1))
if (newOffsets != null) {
- newOffsets.foreach(no => {
- logCheckpoint(no._1, no._2, "New offset to be written for taskname " + no._1)
- writeNewCheckpoint(no._1, no._2)
- info("Ok, new checkpoint has been written for taskname " + no._1)
- })
+ newOffsets.foreach {
+ case (taskName: TaskName, offsets: Map[SystemStreamPartition, String]) => {
+ logCheckpoint(taskName, offsets, "New offset to be written for task: " + taskName)
+ val checkpoint = new Checkpoint(offsets.asJava)
+ checkpointManager.writeCheckpoint(taskName, checkpoint)
+ info(s"Updated the checkpoint of the task: $taskName to: $offsets")
+ }
+ }
}
- manager.stop
- coordinatorStreamManager.stop();
- }
-
- /** Load the most recent checkpoint state for all a specified TaskName. */
- def readLastCheckpoint(taskName:TaskName): Map[SystemStreamPartition, String] = {
- Option(manager.readLastCheckpoint(taskName))
- .getOrElse(new Checkpoint(new java.util.HashMap[SystemStreamPartition, String]()))
- .getOffsets
- .asScala
- .toMap
- }
-
- /**
- * Store a new checkpoint state for specified TaskName, overwriting any previous
- * checkpoint for that TaskName
- */
- def writeNewCheckpoint(tn: TaskName, newOffsets: Map[SystemStreamPartition, String]) {
- val checkpoint = new Checkpoint(newOffsets.asJava)
- manager.writeCheckpoint(tn, checkpoint)
+ checkpointManager.stop()
+ coordinatorStreamManager.stop()
}
def logCheckpoint(tn: TaskName, checkpoint: Map[SystemStreamPartition, String], prefix: String) {
def logLine(tn:TaskName, ssp:SystemStreamPartition, offset:String) = (prefix + ": " + CheckpointTool.SSP_PATTERN + " = %s") format (tn.toString, ssp.getSystem, ssp.getStream, ssp.getPartition.getPartitionId, offset)
- checkpoint.keys.toList.sorted.foreach(ssp => info(logLine(tn, ssp, checkpoint.get(ssp).get)))
+ checkpoint.keys.toList.sorted.foreach(ssp => info(logLine(tn, ssp, checkpoint(ssp))))
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/5ce320c2/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
index 4bb7adf..dadccd9 100644
--- a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
+++ b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
@@ -23,31 +23,45 @@ import java.util
import org.apache.samza.Partition
import org.apache.samza.checkpoint.CheckpointTool.CheckpointToolCommandLine
+import org.apache.samza.checkpoint.CheckpointTool.TaskNameToCheckpointMap
import org.apache.samza.container.TaskName
-import org.apache.samza.checkpoint.TestCheckpointTool.{MockCheckpointManagerFactory, MockSystemFactory}
-import org.apache.samza.config.{Config, MapConfig, SystemConfig, TaskConfig}
+import org.apache.samza.checkpoint.TestCheckpointTool.MockCheckpointManagerFactory
+import org.apache.samza.checkpoint.TestCheckpointTool.MockSystemFactory
+import org.apache.samza.config.Config
+import org.apache.samza.config.MapConfig
+import org.apache.samza.config.SystemConfig
+import org.apache.samza.config.TaskConfig
import org.apache.samza.metrics.MetricsRegistry
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
-import org.apache.samza.system.{SystemAdmin, SystemConsumer, SystemFactory, SystemProducer, SystemStreamMetadata, SystemStreamPartition}
-import org.junit.{Before, Test}
+import org.apache.samza.system.SystemAdmin
+import org.apache.samza.system.SystemConsumer
+import org.apache.samza.system.SystemFactory
+import org.apache.samza.system.SystemProducer
+import org.apache.samza.system.SystemStreamMetadata
+import org.apache.samza.system.SystemStreamPartition
+import org.junit.Before
+import org.junit.Test
import org.mockito.Matchers._
import org.mockito.Mockito._
import org.scalatest.junit.AssertionsForJUnit
import org.scalatest.mockito.MockitoSugar
import scala.collection.JavaConverters._
import org.apache.samza.config.JobConfig
+import org.apache.samza.coordinator.stream.CoordinatorStreamManager
import org.apache.samza.coordinator.stream.MockCoordinatorStreamSystemFactory
-import scala.collection.immutable.HashMap
-
object TestCheckpointTool {
- var checkpointManager: CheckpointManager = null
- var systemConsumer: SystemConsumer = null
- var systemProducer: SystemProducer = null
- var systemAdmin: SystemAdmin = null
+ var checkpointManager: CheckpointManager = _
+ var systemConsumer: SystemConsumer = _
+ var systemProducer: SystemProducer = _
+ var systemAdmin: SystemAdmin = _
+ var coordinatorConfig: Config = new MapConfig()
class MockCheckpointManagerFactory extends CheckpointManagerFactory {
- def getCheckpointManager(config: Config, registry: MetricsRegistry) = checkpointManager
+ def getCheckpointManager(config: Config, registry: MetricsRegistry): CheckpointManager = {
+ coordinatorConfig = config
+ checkpointManager
+ }
}
class MockSystemFactory extends SystemFactory {
@@ -58,7 +72,7 @@ object TestCheckpointTool {
}
class TestCheckpointTool extends AssertionsForJUnit with MockitoSugar {
- var config: MapConfig = null
+ var config: MapConfig = _
val tn0 = new TaskName("Partition 0")
val tn1 = new TaskName("Partition 1")
@@ -66,7 +80,7 @@ class TestCheckpointTool extends AssertionsForJUnit with MockitoSugar {
val p1 = new Partition(1)
@Before
- def setup {
+ def setup() {
config = new MapConfig(Map(
JobConfig.JOB_NAME -> "test",
JobConfig.JOB_COORDINATOR_SYSTEM -> "coordinator",
@@ -91,21 +105,21 @@ class TestCheckpointTool extends AssertionsForJUnit with MockitoSugar {
}
@Test
- def testReadLatestCheckpoint {
+ def testReadLatestCheckpoint() {
val checkpointTool = CheckpointTool(config, null)
- checkpointTool.run
+ checkpointTool.run()
verify(TestCheckpointTool.checkpointManager).readLastCheckpoint(tn0)
verify(TestCheckpointTool.checkpointManager).readLastCheckpoint(tn1)
verify(TestCheckpointTool.checkpointManager, never()).writeCheckpoint(any(), any())
}
@Test
- def testOverwriteCheckpoint {
+ def testOverwriteCheckpoint() {
val toOverwrite = Map(tn0 -> Map(new SystemStreamPartition("test", "foo", p0) -> "42"),
tn1 -> Map(new SystemStreamPartition("test", "foo", p1) -> "43"))
val checkpointTool = CheckpointTool(config, toOverwrite)
- checkpointTool.run
+ checkpointTool.run()
verify(TestCheckpointTool.checkpointManager)
.writeCheckpoint(tn0, new Checkpoint(Map(new SystemStreamPartition("test", "foo", p0) -> "42").asJava))
verify(TestCheckpointTool.checkpointManager)
@@ -127,4 +141,32 @@ class TestCheckpointTool extends AssertionsForJUnit with MockitoSugar {
assert(result(new TaskName("Partition 0")).size == 2)
assert(result(new TaskName("Partition 1")).size == 3)
}
+
+ @Test
+ def testShouldInstantiateCheckpointManagerWithConfigurationFromCoordinatorStream(): Unit = {
+ val offsetMap: TaskNameToCheckpointMap = Map(tn0 -> Map(new SystemStreamPartition("test", "foo", p0) -> "42"),
+ tn1 -> Map(new SystemStreamPartition("test", "foo", p1) -> "43"))
+ val coordinatorStreamManager = mock[CoordinatorStreamManager]
+ val userDefinedConfig: Config = new MapConfig(Map(
+ JobConfig.JOB_NAME -> "test",
+ JobConfig.JOB_COORDINATOR_SYSTEM -> "coordinator",
+ TaskConfig.INPUT_STREAMS -> "test.foo",
+ TaskConfig.CHECKPOINT_MANAGER_FACTORY -> classOf[MockCheckpointManagerFactory].getName,
+ SystemConfig.SYSTEM_FACTORY.format("test") -> classOf[MockSystemFactory].getName,
+ SystemConfig.SYSTEM_FACTORY.format("coordinator") -> classOf[MockCoordinatorStreamSystemFactory].getName,
+ TaskConfig.GROUPER_FACTORY -> "org.apache.samza.container.grouper.task.GroupByContainerCountFactory"
+ ).asJava)
+
+ when(coordinatorStreamManager.getConfig).thenReturn(userDefinedConfig)
+
+ val checkpointTool: CheckpointTool = new CheckpointTool(offsetMap, coordinatorStreamManager)
+ checkpointTool.run()
+
+ verify(TestCheckpointTool.checkpointManager)
+ .writeCheckpoint(tn0, new Checkpoint(Map(new SystemStreamPartition("test", "foo", p0) -> "42").asJava))
+ verify(TestCheckpointTool.checkpointManager)
+ .writeCheckpoint(tn1, new Checkpoint(Map(new SystemStreamPartition("test", "foo", p1) -> "43").asJava))
+ verify(coordinatorStreamManager).getConfig
+ assert(TestCheckpointTool.coordinatorConfig == userDefinedConfig)
+ }
}