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/04/29 04:57:49 UTC
[2/4] samza git commit: SAMZA-465: Use coordinator stream and
eliminate CheckpointManager
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala b/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
deleted file mode 100644
index 10ff1f4..0000000
--- a/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.checkpoint.file
-
-import java.io.File
-import scala.collection.JavaConversions._
-import java.util.Random
-import org.junit.Assert._
-import org.junit.{After, Before, Test}
-import org.apache.samza.SamzaException
-import org.apache.samza.Partition
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.system.SystemStreamPartition
-import org.apache.samza.container.TaskName
-import org.junit.rules.TemporaryFolder
-
-class TestFileSystemCheckpointManager {
- val checkpointRoot = System.getProperty("java.io.tmpdir") // TODO: Move this out of tmp, into our build dir
- val taskName = new TaskName("Warwickshire")
- val baseFileLocation = new File(checkpointRoot)
-
- val tempFolder = new TemporaryFolder
-
- @Before
- def createTempFolder = tempFolder.create()
-
- @After
- def deleteTempFolder = tempFolder.delete()
-
- @Test
- def testReadForCheckpointFileThatDoesNotExistShouldReturnNull {
- val cpm = new FileSystemCheckpointManager("some-job-name", tempFolder.getRoot)
- assertNull(cpm.readLastCheckpoint(taskName))
- }
-
- @Test
- def testReadForCheckpointFileThatDoesExistShouldReturnProperCheckpoint {
- val cp = new Checkpoint(Map(
- new SystemStreamPartition("a", "b", new Partition(0)) -> "c",
- new SystemStreamPartition("a", "c", new Partition(1)) -> "d",
- new SystemStreamPartition("b", "d", new Partition(2)) -> "e"))
-
- var readCp:Checkpoint = null
- val cpm = new FileSystemCheckpointManager("some-job-name", tempFolder.getRoot)
-
- cpm.start
- cpm.writeCheckpoint(taskName, cp)
- readCp = cpm.readLastCheckpoint(taskName)
- cpm.stop
-
- assertNotNull(readCp)
- cp.equals(readCp)
- assertEquals(cp.getOffsets.keySet(), readCp.getOffsets.keySet())
- assertEquals(cp.getOffsets, readCp.getOffsets)
- assertEquals(cp, readCp)
- }
-
- @Test
- def testMissingRootDirectoryShouldFailOnManagerCreation {
- val cpm = new FileSystemCheckpointManager("some-job-name", new File(checkpointRoot + new Random().nextInt))
- try {
- cpm.start
- fail("Expected an exception since root directory for fs checkpoint manager doesn't exist.")
- } catch {
- case e: SamzaException => None // this is expected
- }
- cpm.stop
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
index cab31ca..58d7fe8 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
@@ -19,6 +19,7 @@
package org.apache.samza.container
+import java.util
import scala.collection.JavaConversions._
import org.apache.samza.Partition
import org.apache.samza.config.Config
@@ -29,7 +30,6 @@ import org.apache.samza.coordinator.server.JobServlet
import org.apache.samza.job.model.ContainerModel
import org.apache.samza.job.model.JobModel
import org.apache.samza.job.model.TaskModel
-import org.apache.samza.metrics.JmxServer
import org.apache.samza.serializers.SerdeManager
import org.apache.samza.system.IncomingMessageEnvelope
import org.apache.samza.system.StreamMetadataCache
@@ -54,21 +54,25 @@ import org.scalatest.junit.AssertionsForJUnit
import java.lang.Thread.UncaughtExceptionHandler
import org.apache.samza.serializers._
import org.apache.samza.SamzaException
+import org.apache.samza.checkpoint.CheckpointManager
class TestSamzaContainer extends AssertionsForJUnit {
@Test
def testReadJobModel {
val config = new MapConfig(Map("a" -> "b"))
+ val offsets = new util.HashMap[SystemStreamPartition, String]();
+ offsets.put(new SystemStreamPartition("system","stream", new Partition(0)), "1")
val tasks = Map(
- new TaskName("t1") -> new TaskModel(new TaskName("t1"), Set[SystemStreamPartition](), new Partition(0)),
- new TaskName("t2") -> new TaskModel(new TaskName("t2"), Set[SystemStreamPartition](), new Partition(0)))
+ new TaskName("t1") -> new TaskModel(new TaskName("t1"), offsets, new Partition(0)),
+ new TaskName("t2") -> new TaskModel(new TaskName("t2"), offsets, new Partition(0)))
val containers = Map(
Integer.valueOf(0) -> new ContainerModel(0, tasks),
Integer.valueOf(1) -> new ContainerModel(1, tasks))
val jobModel = new JobModel(config, containers)
+ def jobModelGenerator(): JobModel = jobModel
val server = new HttpServer
- val coordinator = new JobCoordinator(jobModel, server)
- coordinator.server.addServlet("/*", new JobServlet(jobModel))
+ val coordinator = new JobCoordinator(jobModel, server, new MockCheckpointManager)
+ coordinator.server.addServlet("/*", new JobServlet(jobModelGenerator))
try {
coordinator.start
assertEquals(jobModel, SamzaContainer.readJobModel(server.getUrl.toString))
@@ -180,7 +184,7 @@ class TestSamzaContainer extends AssertionsForJUnit {
val config = new MapConfig
assertTrue(defaultSerdesFromSerdeName("byte", "testSystemException", config).isInstanceOf[ByteSerde])
assertTrue(defaultSerdesFromSerdeName("integer", "testSystemException", config).isInstanceOf[IntegerSerde])
- assertTrue(defaultSerdesFromSerdeName("json", "testSystemException", config).isInstanceOf[JsonSerde])
+ assertTrue(defaultSerdesFromSerdeName("json", "testSystemException", config).isInstanceOf[JsonSerde[Object]])
assertTrue(defaultSerdesFromSerdeName("long", "testSystemException", config).isInstanceOf[LongSerde])
assertTrue(defaultSerdesFromSerdeName("serializable", "testSystemException", config).isInstanceOf[SerializableSerde[java.io.Serializable @unchecked]])
assertTrue(defaultSerdesFromSerdeName("string", "testSystemException", config).isInstanceOf[StringSerde])
@@ -196,3 +200,8 @@ class TestSamzaContainer extends AssertionsForJUnit {
assertTrue(throwSamzaException)
}
}
+
+class MockCheckpointManager extends CheckpointManager(null, null) {
+ override def start() = {}
+ override def stop() = {}
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/container/grouper/task/TestGroupByContainerCount.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/grouper/task/TestGroupByContainerCount.scala b/samza-core/src/test/scala/org/apache/samza/container/grouper/task/TestGroupByContainerCount.scala
index 1ee5c06..2c7cb28 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/grouper/task/TestGroupByContainerCount.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/grouper/task/TestGroupByContainerCount.scala
@@ -68,6 +68,6 @@ class TestGroupByContainerCount {
}
private def getTaskModel(name: String, partitionId: Int) = {
- new TaskModel(new TaskName(name), Set[SystemStreamPartition](), new Partition(partitionId))
+ new TaskModel(new TaskName(name), Map[SystemStreamPartition, String](), new Partition(partitionId))
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/coordinator/TestJobCoordinator.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/coordinator/TestJobCoordinator.scala b/samza-core/src/test/scala/org/apache/samza/coordinator/TestJobCoordinator.scala
index a8e5d36..d9ae187 100644
--- a/samza-core/src/test/scala/org/apache/samza/coordinator/TestJobCoordinator.scala
+++ b/samza-core/src/test/scala/org/apache/samza/coordinator/TestJobCoordinator.scala
@@ -19,17 +19,19 @@
package org.apache.samza.coordinator
-import org.junit.Test
+
+import java.net.SocketTimeoutException
+
+import org.apache.samza.util.Util
+import org.junit.{After, Test}
import org.junit.Assert._
+import org.junit.rules.ExpectedException
import scala.collection.JavaConversions._
import org.apache.samza.config.MapConfig
import org.apache.samza.config.TaskConfig
import org.apache.samza.config.SystemConfig
-import org.apache.samza.container.TaskName
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.checkpoint.CheckpointManagerFactory
-import org.apache.samza.checkpoint.CheckpointManager
-import org.apache.samza.metrics.MetricsRegistry
+import org.apache.samza.container.{SamzaContainer, TaskName}
+import org.apache.samza.metrics.{MetricsRegistryMap, MetricsRegistry}
import org.apache.samza.config.Config
import org.apache.samza.system.SystemFactory
import org.apache.samza.system.SystemAdmin
@@ -40,67 +42,209 @@ import org.apache.samza.Partition
import org.apache.samza.job.model.JobModel
import org.apache.samza.job.model.ContainerModel
import org.apache.samza.job.model.TaskModel
+import org.apache.samza.config.JobConfig
+import org.apache.samza.system.IncomingMessageEnvelope
+import org.apache.samza.system.SystemConsumer
+import org.apache.samza.coordinator.stream.{MockCoordinatorStreamWrappedConsumer, MockCoordinatorStreamSystemFactory}
class TestJobCoordinator {
/**
- * Builds a coordinator from config, and then compares it with what was
- * expected. We simulate having a checkpoint manager that has 2 task
- * changelog entries, and our model adds a third task. Expectation is that
- * the JobCoordinator will assign the new task with a new changelog
- * partition.
+ * Builds a coordinator from config, and then compares it with what was
+ * expected. We simulate having a checkpoint manager that has 2 task
+ * changelog entries, and our model adds a third task. Expectation is that
+ * the JobCoordinator will assign the new task with a new changelog
+ * partition
*/
@Test
def testJobCoordinator {
- val containerCount = 2
- val config = new MapConfig(Map(
- TaskConfig.CHECKPOINT_MANAGER_FACTORY -> classOf[MockCheckpointManagerFactory].getCanonicalName,
- TaskConfig.INPUT_STREAMS -> "test.stream1",
- (SystemConfig.SYSTEM_FACTORY format "test") -> classOf[MockSystemFactory].getCanonicalName))
- val coordinator = JobCoordinator(config, containerCount)
+ val task0Name = new TaskName("Partition 0")
+ val checkpoint0 = Map(new SystemStreamPartition("test", "stream1", new Partition(0)) -> "4")
+ val task1Name = new TaskName("Partition 1")
+ val checkpoint1 = Map(new SystemStreamPartition("test", "stream1", new Partition(1)) -> "3")
+ val task2Name = new TaskName("Partition 2")
+ val checkpoint2 = Map(new SystemStreamPartition("test", "stream1", new Partition(2)) -> null)
- // Construct the expected JobModel, so we can compare it to
+ // Construct the expected JobModel, so we can compare it to
// JobCoordinator's JobModel.
+ val container0Tasks = Map(
+ task0Name -> new TaskModel(task0Name, checkpoint0, new Partition(4)),
+ task2Name -> new TaskModel(task2Name, checkpoint2, new Partition(5)))
+ val container1Tasks = Map(
+ task1Name -> new TaskModel(task1Name, checkpoint1, new Partition(3)))
+ val containers = Map(
+ Integer.valueOf(0) -> new ContainerModel(0, container0Tasks),
+ Integer.valueOf(1) -> new ContainerModel(1, container1Tasks))
+
+
+ // The test does not pass offsets for task2 (Partition 2) to the checkpointmanager, this will verify that we get an offset 0 for this partition
+ val checkpointOffset0 = MockCoordinatorStreamWrappedConsumer.CHECKPOINTPREFIX + "mock:" +
+ task0Name.getTaskName() -> (Util.sspToString(checkpoint0.keySet.iterator.next()) + ":" + checkpoint0.values.iterator.next())
+ val checkpointOffset1 = MockCoordinatorStreamWrappedConsumer.CHECKPOINTPREFIX + "mock:" +
+ task1Name.getTaskName() -> (Util.sspToString(checkpoint1.keySet.iterator.next()) + ":" + checkpoint1.values.iterator.next())
+ val changelogInfo0 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task0Name.getTaskName() -> "4"
+ val changelogInfo1 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task1Name.getTaskName() -> "3"
+ val changelogInfo2 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task2Name.getTaskName() -> "5"
+
+ // Configs which are processed by the MockCoordinatorStream as special configs which are interpreted as
+ // SetCheckpoint and SetChangelog
+ val otherConfigs = Map(
+ checkpointOffset0,
+ checkpointOffset1,
+ changelogInfo0,
+ changelogInfo1,
+ changelogInfo2
+ )
+
+ val config = Map(
+ JobConfig.JOB_NAME -> "test",
+ JobConfig.JOB_COORDINATOR_SYSTEM -> "coordinator",
+ JobConfig.JOB_CONTAINER_COUNT -> "2",
+ TaskConfig.INPUT_STREAMS -> "test.stream1",
+ SystemConfig.SYSTEM_FACTORY.format("test") -> classOf[MockSystemFactory].getCanonicalName,
+ SystemConfig.SYSTEM_FACTORY.format("coordinator") -> classOf[MockCoordinatorStreamSystemFactory].getName
+ )
+
+ // We want the mocksystemconsumer to use the same instance across runs
+ MockCoordinatorStreamSystemFactory.enableMockConsumerCache()
+
+ val coordinator = JobCoordinator(new MapConfig(config ++ otherConfigs))
+ coordinator.start
+ val jobModel = new JobModel(new MapConfig(config), containers)
+ assertEquals(new MapConfig(config), coordinator.jobModel.getConfig)
+ assertEquals(jobModel, coordinator.jobModel)
+ }
+
+ @Test
+ def testJobCoordinatorCheckpointing = {
+ System.out.println("test ")
val task0Name = new TaskName("Partition 0")
+ val checkpoint0 = Map(new SystemStreamPartition("test", "stream1", new Partition(0)) -> "4")
val task1Name = new TaskName("Partition 1")
+ val checkpoint1 = Map(new SystemStreamPartition("test", "stream1", new Partition(1)) ->"3")
val task2Name = new TaskName("Partition 2")
+ val checkpoint2 = Map(new SystemStreamPartition("test", "stream1", new Partition(2)) -> "8")
+
+ // Construct the expected JobModel, so we can compare it to
+ // JobCoordinator's JobModel.
val container0Tasks = Map(
- task0Name -> new TaskModel(task0Name, Set(new SystemStreamPartition("test", "stream1", new Partition(0))), new Partition(4)),
- task2Name -> new TaskModel(task2Name, Set(new SystemStreamPartition("test", "stream1", new Partition(2))), new Partition(5)))
+ task0Name -> new TaskModel(task0Name, checkpoint0, new Partition(4)),
+ task2Name -> new TaskModel(task2Name, checkpoint2, new Partition(5)))
val container1Tasks = Map(
- task1Name -> new TaskModel(task1Name, Set(new SystemStreamPartition("test", "stream1", new Partition(1))), new Partition(3)))
+ task1Name -> new TaskModel(task1Name, checkpoint1, new Partition(3)))
val containers = Map(
Integer.valueOf(0) -> new ContainerModel(0, container0Tasks),
Integer.valueOf(1) -> new ContainerModel(1, container1Tasks))
- val jobModel = new JobModel(config, containers)
- assertEquals(config, coordinator.jobModel.getConfig)
- assertEquals(jobModel, coordinator.jobModel)
+
+
+ // The test does not pass offsets for task2 (Partition 2) to the checkpointmanager, this will verify that we get an offset 0 for this partition
+ val checkpointOffset0 = MockCoordinatorStreamWrappedConsumer.CHECKPOINTPREFIX + "mock:" +
+ task0Name.getTaskName() -> (Util.sspToString(checkpoint0.keySet.iterator.next()) + ":" + checkpoint0.values.iterator.next())
+ val checkpointOffset1 = MockCoordinatorStreamWrappedConsumer.CHECKPOINTPREFIX + "mock:" +
+ task1Name.getTaskName() -> (Util.sspToString(checkpoint1.keySet.iterator.next()) + ":" + checkpoint1.values.iterator.next())
+ val checkpointOffset2 = MockCoordinatorStreamWrappedConsumer.CHECKPOINTPREFIX + "mock:" +
+ task2Name.getTaskName() -> (Util.sspToString(checkpoint2.keySet.iterator.next()) + ":" + checkpoint2.values.iterator.next())
+ val changelogInfo0 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task0Name.getTaskName() -> "4"
+ val changelogInfo1 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task1Name.getTaskName() -> "3"
+ val changelogInfo2 = MockCoordinatorStreamWrappedConsumer.CHANGELOGPREFIX + "mock:" + task2Name.getTaskName() -> "5"
+
+ // Configs which are processed by the MockCoordinatorStream as special configs which are interpreted as
+ // SetCheckpoint and SetChangelog
+ // Write a couple of checkpoints that the job coordinator will process
+ val otherConfigs = Map(
+ checkpointOffset0,
+ changelogInfo0
+ )
+
+ val config = Map(
+ JobConfig.JOB_NAME -> "test",
+ JobConfig.JOB_COORDINATOR_SYSTEM -> "coordinator",
+ JobConfig.JOB_CONTAINER_COUNT -> "2",
+ TaskConfig.INPUT_STREAMS -> "test.stream1",
+ SystemConfig.SYSTEM_FACTORY.format("test") -> classOf[MockSystemFactory].getCanonicalName,
+ SystemConfig.SYSTEM_FACTORY.format("coordinator") -> classOf[MockCoordinatorStreamSystemFactory].getName
+ )
+
+ // Enable caching on MockConsumer to add more messages later
+ MockCoordinatorStreamSystemFactory.enableMockConsumerCache()
+
+ // start the job coordinator and verify if it has all the checkpoints through http port
+ val coordinator = JobCoordinator(new MapConfig(config ++ otherConfigs))
+ coordinator.start
+ val url = coordinator.server.getUrl.toString
+
+ // Verify if the jobCoordinator has seen the checkpoints
+ var offsets = extractOffsetsFromJobCoordinator(url)
+ assertEquals(1, offsets.size)
+ assertEquals(checkpoint0.head._2, offsets.getOrElse(checkpoint0.head._1, fail()))
+
+ // Write more checkpoints
+ val wrappedConsumer = new MockCoordinatorStreamSystemFactory()
+ .getConsumer(null, null, null)
+ .asInstanceOf[MockCoordinatorStreamWrappedConsumer]
+
+ var moreMessageConfigs = Map(
+ checkpointOffset1
+ )
+
+ wrappedConsumer.addMoreMessages(new MapConfig(moreMessageConfigs))
+
+ // Verify if the coordinator has seen it
+ offsets = extractOffsetsFromJobCoordinator(url)
+ assertEquals(2, offsets.size)
+ assertEquals(checkpoint0.head._2, offsets.getOrElse(checkpoint0.head._1, fail()))
+ assertEquals(checkpoint1.head._2, offsets.getOrElse(checkpoint1.head._1, fail()))
+
+ // Write more checkpoints but block on read on the mock consumer
+ moreMessageConfigs = Map(
+ checkpointOffset2
+ )
+
+ wrappedConsumer.addMoreMessages(new MapConfig(moreMessageConfigs))
+
+ // Simulate consumer being blocked (Job coordinator waiting to read new checkpoints from coordinator after container failure)
+ val latch = wrappedConsumer.blockPool();
+
+ // verify if the port times out
+ var seenException = false
+ try {
+ extractOffsetsFromJobCoordinator(url)
+ }
+ catch {
+ case se: SocketTimeoutException => seenException = true
+ }
+ assertTrue(seenException)
+
+ // verify if it has read the new checkpoints after job coordinator has loaded the new checkpoints
+ latch.countDown()
+ offsets = extractOffsetsFromJobCoordinator(url)
+ assertEquals(offsets.size, 3)
+ assertEquals(checkpoint0.head._2, offsets.getOrElse(checkpoint0.head._1, fail()))
+ assertEquals(checkpoint1.head._2, offsets.getOrElse(checkpoint1.head._1, fail()))
+ assertEquals(checkpoint2.head._2, offsets.getOrElse(checkpoint2.head._1, fail()))
+ coordinator.stop
}
-}
-object MockCheckpointManager {
- var mapping: java.util.Map[TaskName, java.lang.Integer] = Map[TaskName, java.lang.Integer](
- new TaskName("Partition 0") -> 4,
- new TaskName("Partition 1") -> 3)
-}
+ def extractOffsetsFromJobCoordinator(url : String) = {
+ val jobModel = SamzaContainer.readJobModel(url.toString)
+ val taskModels = jobModel.getContainers.values().flatMap(_.getTasks.values())
+ val offsets = taskModels.flatMap(_.getCheckpointedOffsets).toMap
+ offsets.filter(_._2 != null)
+ }
-class MockCheckpointManagerFactory extends CheckpointManagerFactory {
- def getCheckpointManager(config: Config, registry: MetricsRegistry) = new MockCheckpointManager
-}
-class MockCheckpointManager extends CheckpointManager {
- def start() {}
- def register(taskName: TaskName) {}
- def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {}
- def readLastCheckpoint(taskName: TaskName) = null
- def readChangeLogPartitionMapping = MockCheckpointManager.mapping
- def writeChangeLogPartitionMapping(mapping: java.util.Map[TaskName, java.lang.Integer]) {
- MockCheckpointManager.mapping = mapping
+ @After
+ def tearDown() = {
+ MockCoordinatorStreamSystemFactory.disableMockConsumerCache()
}
- def stop() {}
}
class MockSystemFactory extends SystemFactory {
- def getConsumer(systemName: String, config: Config, registry: MetricsRegistry) = null
+ def getConsumer(systemName: String, config: Config, registry: MetricsRegistry) = new SystemConsumer {
+ def start() {}
+ def stop() {}
+ def register(systemStreamPartition: SystemStreamPartition, offset: String) {}
+ def poll(systemStreamPartitions: java.util.Set[SystemStreamPartition], timeout: Long) = new java.util.HashMap[SystemStreamPartition, java.util.List[IncomingMessageEnvelope]]()
+ }
def getProducer(systemName: String, config: Config, registry: MetricsRegistry) = null
def getAdmin(systemName: String, config: Config) = new MockSystemAdmin
}
@@ -117,5 +261,11 @@ class MockSystemAdmin extends SystemAdmin {
Map(streamNames.toList.head -> new SystemStreamMetadata("foo", partitionMetadata))
}
- override def createChangelogStream(streamName: String, numOfPartitions: Int) = ???
+ override def createChangelogStream(topicName: String, numOfChangeLogPartitions: Int) {
+ new UnsupportedOperationException("Method not implemented.")
+ }
+
+ override def createCoordinatorStream(streamName: String) {
+ new UnsupportedOperationException("Method not implemented.")
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/coordinator/server/TestHttpServer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/coordinator/server/TestHttpServer.scala b/samza-core/src/test/scala/org/apache/samza/coordinator/server/TestHttpServer.scala
index 9a8e145..6ce8aa9 100644
--- a/samza-core/src/test/scala/org/apache/samza/coordinator/server/TestHttpServer.scala
+++ b/samza-core/src/test/scala/org/apache/samza/coordinator/server/TestHttpServer.scala
@@ -19,9 +19,9 @@
package org.apache.samza.coordinator.server
+import org.apache.samza.util.Util
import org.junit.Assert._
import org.junit.Test
-import org.apache.samza.util.Util
import java.net.URL
class TestHttpServer {
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala b/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala
index 3a710a8..a1efe6f 100644
--- a/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala
+++ b/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala
@@ -56,7 +56,7 @@ class TestProcessJob {
}
}
-class MockJobCoordinator extends JobCoordinator(null, null) {
+class MockJobCoordinator extends JobCoordinator(null, null, null) {
var stopped: Boolean = false
override def start: Unit = { }
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-core/src/test/scala/org/apache/samza/serializers/TestJsonSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestJsonSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestJsonSerde.scala
index 6046071..4f1c14c 100644
--- a/samza-core/src/test/scala/org/apache/samza/serializers/TestJsonSerde.scala
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestJsonSerde.scala
@@ -23,11 +23,13 @@ import org.junit.Assert._
import org.junit.Test
import scala.collection.JavaConversions._
+import scala.collection.immutable.HashMap
+
class TestJsonSerde {
@Test
def testJsonSerdeShouldWork {
- val serde = new JsonSerde
+ val serde = new JsonSerde[java.util.HashMap[String, Object]]
val obj = new java.util.HashMap[String, Object](Map[String, Object]("hi" -> "bye", "why" -> new java.lang.Integer(2)))
val bytes = serde.toBytes(obj)
assertEquals(obj, serde.fromBytes(bytes))
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
deleted file mode 100644
index 5d8ee4f..0000000
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.checkpoint.kafka
-
-import java.util
-import org.apache.samza.SamzaException
-import org.apache.samza.container.TaskName
-import org.codehaus.jackson.`type`.TypeReference
-import org.codehaus.jackson.map.ObjectMapper
-import scala.collection.JavaConversions._
-
-/**
- * Kafka Checkpoint Log-specific key used to identify what type of entry is
- * written for any particular log entry.
- *
- * @param map Backing map to hold key values
- */
-class KafkaCheckpointLogKey private (val map: Map[String, String]) {
- // This might be better as a case class...
- import KafkaCheckpointLogKey._
-
- /**
- * Serialize this key to bytes
- * @return Key as bytes
- */
- def toBytes(): Array[Byte] = {
- val jMap = new util.HashMap[String, String](map.size)
- jMap.putAll(map)
-
- JSON_MAPPER.writeValueAsBytes(jMap)
- }
-
- private def getKey = map.getOrElse(CHECKPOINT_KEY_KEY, throw new SamzaException("No " + CHECKPOINT_KEY_KEY + " in map for Kafka Checkpoint log key"))
-
- /**
- * Is this key for a checkpoint entry?
- *
- * @return true iff this key's entry is for a checkpoint
- */
- def isCheckpointKey = getKey.equals(CHECKPOINT_KEY_TYPE)
-
- /**
- * Is this key for a changelog partition mapping?
- *
- * @return true iff this key's entry is for a changelog partition mapping
- */
- def isChangelogPartitionMapping = getKey.equals(CHANGELOG_PARTITION_KEY_TYPE)
-
- /**
- * If this Key is for a checkpoint entry, return its associated TaskName.
- *
- * @return TaskName for this checkpoint or throw an exception if this key does not have a TaskName entry
- */
- def getCheckpointTaskName = {
- val asString = map.getOrElse(CHECKPOINT_TASKNAME_KEY, throw new SamzaException("No TaskName in checkpoint key: " + this))
- new TaskName(asString)
- }
-
- def canEqual(other: Any): Boolean = other.isInstanceOf[KafkaCheckpointLogKey]
-
- override def equals(other: Any): Boolean = other match {
- case that: KafkaCheckpointLogKey =>
- (that canEqual this) &&
- map == that.map
- case _ => false
- }
-
- override def hashCode(): Int = {
- val state = Seq(map)
- state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
- }
-}
-
-object KafkaCheckpointLogKey {
- /**
- * Messages in the checkpoint log have keys associated with them. These keys are maps that describe the message's
- * type, either a checkpoint or a changelog-partition-mapping.
- */
- val CHECKPOINT_KEY_KEY = "type"
- val CHECKPOINT_KEY_TYPE = "checkpoint"
- val CHANGELOG_PARTITION_KEY_TYPE = "changelog-partition-mapping"
- val CHECKPOINT_TASKNAME_KEY = "taskName"
- val SYSTEMSTREAMPARTITION_GROUPER_FACTORY_KEY = "systemstreampartition-grouper-factory"
-
- /**
- * Partition mapping keys have no dynamic values, so we just need one instance.
- */
- val CHANGELOG_PARTITION_MAPPING_KEY = new KafkaCheckpointLogKey(Map(CHECKPOINT_KEY_KEY -> CHANGELOG_PARTITION_KEY_TYPE))
-
- private val JSON_MAPPER = new ObjectMapper()
- val KEY_TYPEREFERENCE = new TypeReference[util.HashMap[String, String]]() {}
-
- var systemStreamPartitionGrouperFactoryString:Option[String] = None
-
- /**
- * Set the name of the factory configured to provide the SystemStreamPartition grouping
- * so it be included in the key.
- *
- * @param str Config value of SystemStreamPartition Grouper Factory
- */
- def setSystemStreamPartitionGrouperFactoryString(str:String) = {
- systemStreamPartitionGrouperFactoryString = Some(str)
- }
-
- /**
- * Get the name of the factory configured to provide the SystemStreamPartition grouping
- * so it be included in the key
- */
- def getSystemStreamPartitionGrouperFactoryString = systemStreamPartitionGrouperFactoryString.getOrElse(throw new SamzaException("No SystemStreamPartition grouping factory string has been set."))
-
- /**
- * Build a key for a a checkpoint log entry for a particular TaskName
- * @param taskName TaskName to build for this checkpoint entry
- *
- * @return Key for checkpoint log entry
- */
- def getCheckpointKey(taskName:TaskName) = {
- val map = Map(CHECKPOINT_KEY_KEY -> CHECKPOINT_KEY_TYPE,
- CHECKPOINT_TASKNAME_KEY -> taskName.getTaskName,
- SYSTEMSTREAMPARTITION_GROUPER_FACTORY_KEY -> getSystemStreamPartitionGrouperFactoryString)
-
- new KafkaCheckpointLogKey(map)
- }
-
- /**
- * Build a key for a changelog partition mapping entry
- *
- * @return Key for changelog partition mapping entry
- */
- def getChangelogPartitionMappingKey() = CHANGELOG_PARTITION_MAPPING_KEY
-
- /**
- * Deserialize a Kafka checkpoint log key
- * @param bytes Serialized (via JSON) Kafka checkpoint log key
- * @return Checkpoint log key
- */
- def fromBytes(bytes: Array[Byte]): KafkaCheckpointLogKey = {
- try {
- val jmap: util.HashMap[String, String] = JSON_MAPPER.readValue(bytes, KEY_TYPEREFERENCE)
-
- if(!jmap.containsKey(CHECKPOINT_KEY_KEY)) {
- throw new SamzaException("No type entry in checkpoint key: " + jmap)
- }
-
- // Only checkpoint keys have ssp grouper factory keys
- if(jmap.get(CHECKPOINT_KEY_KEY).equals(CHECKPOINT_KEY_TYPE)) {
- val sspGrouperFactory = jmap.get(SYSTEMSTREAMPARTITION_GROUPER_FACTORY_KEY)
-
- if (sspGrouperFactory == null) {
- throw new SamzaException("No SystemStreamPartition Grouper factory entry in checkpoint key: " + jmap)
- }
-
- if (!sspGrouperFactory.equals(getSystemStreamPartitionGrouperFactoryString)) {
- throw new DifferingSystemStreamPartitionGrouperFactoryValues(sspGrouperFactory, getSystemStreamPartitionGrouperFactoryString)
- }
- }
-
- new KafkaCheckpointLogKey(jmap.toMap)
- } catch {
- case e: Exception =>
- throw new SamzaException("Exception while deserializing checkpoint key", e)
- }
- }
-}
-
-class DifferingSystemStreamPartitionGrouperFactoryValues(inKey:String, inConfig:String) extends SamzaException {
- override def getMessage() = "Checkpoint key's SystemStreamPartition Grouper factory (" + inKey +
- ") does not match value from current configuration (" + inConfig + "). " +
- "This likely means the SystemStreamPartitionGrouper was changed between job runs, which is not supported."
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
deleted file mode 100644
index c9504ec..0000000
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
+++ /dev/null
@@ -1,427 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.checkpoint.kafka
-
-import org.apache.samza.util.Logging
-import java.nio.ByteBuffer
-import java.util
-import kafka.admin.AdminUtils
-import kafka.api._
-import kafka.common.ErrorMapping
-import kafka.common.InvalidMessageSizeException
-import kafka.common.TopicAndPartition
-import kafka.common.TopicExistsException
-import kafka.common.UnknownTopicOrPartitionException
-import kafka.consumer.SimpleConsumer
-import kafka.message.InvalidMessageException
-import kafka.utils.Utils
-import org.I0Itec.zkclient.ZkClient
-import org.apache.samza.SamzaException
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.checkpoint.CheckpointManager
-import org.apache.samza.container.TaskName
-import org.apache.samza.serializers.CheckpointSerde
-import org.apache.samza.system.kafka.TopicMetadataCache
-import org.apache.samza.util.ExponentialSleepStrategy
-import org.apache.samza.util.TopicMetadataStore
-import scala.collection.mutable
-import java.util.Properties
-import org.apache.kafka.clients.producer.{Producer, ProducerRecord}
-import org.apache.samza.util.KafkaUtil
-
-/**
- * Kafka checkpoint manager is used to store checkpoints in a Kafka topic.
- * To read a checkpoint for a specific taskName, we find the newest message
- * keyed to that taskName. If there is no such message, no checkpoint data
- * exists. The underlying log has a single partition into which all
- * checkpoints and TaskName to changelog partition mappings are written.
- */
-class KafkaCheckpointManager(
- clientId: String,
- checkpointTopic: String,
- systemName: String,
- replicationFactor: Int,
- socketTimeout: Int,
- bufferSize: Int,
- fetchSize: Int,
- metadataStore: TopicMetadataStore,
- connectProducer: () => Producer[Array[Byte], Array[Byte]],
- connectZk: () => ZkClient,
- systemStreamPartitionGrouperFactoryString: String,
- retryBackoff: ExponentialSleepStrategy = new ExponentialSleepStrategy,
- serde: CheckpointSerde = new CheckpointSerde,
- checkpointTopicProperties: Properties = new Properties) extends CheckpointManager with Logging {
- import KafkaCheckpointManager._
-
- var taskNames = Set[TaskName]()
- var producer: Producer[Array[Byte], Array[Byte]] = null
- var taskNamesToOffsets: Map[TaskName, Checkpoint] = null
-
- var startingOffset: Option[Long] = None // Where to start reading for each subsequent call of readCheckpoint
-
- KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString(systemStreamPartitionGrouperFactoryString)
-
- info("Creating KafkaCheckpointManager with: clientId=%s, checkpointTopic=%s, systemName=%s" format(clientId, checkpointTopic, systemName))
-
- /**
- * Write Checkpoint for specified taskName to log
- *
- * @param taskName Specific Samza taskName of which to write a checkpoint of.
- * @param checkpoint Reference to a Checkpoint object to store offset data in.
- **/
- override def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
- val key = KafkaCheckpointLogKey.getCheckpointKey(taskName)
- val keyBytes = key.toBytes()
- val msgBytes = serde.toBytes(checkpoint)
-
- writeLog(CHECKPOINT_LOG4J_ENTRY, keyBytes, msgBytes)
- }
-
- /**
- * Write the taskName to partition mapping that is being maintained by this CheckpointManager
- *
- * @param changelogPartitionMapping Each TaskName's partition within the changelog
- */
- override def writeChangeLogPartitionMapping(changelogPartitionMapping: util.Map[TaskName, java.lang.Integer]) {
- val key = KafkaCheckpointLogKey.getChangelogPartitionMappingKey()
- val keyBytes = key.toBytes()
- val msgBytes = serde.changelogPartitionMappingToBytes(changelogPartitionMapping)
-
- writeLog(CHANGELOG_PARTITION_MAPPING_LOG4j, keyBytes, msgBytes)
- }
-
- /**
- * Common code for writing either checkpoints or changelog-partition-mappings to the log
- *
- * @param logType Type of entry that is being written, for logging
- * @param key pre-serialized key for message
- * @param msg pre-serialized message to write to log
- */
- private def writeLog(logType:String, key: Array[Byte], msg: Array[Byte]) {
- retryBackoff.run(
- loop => {
- if (producer == null) {
- producer = connectProducer()
- }
-
- producer.send(new ProducerRecord(checkpointTopic, 0, key, msg)).get()
- loop.done
- },
-
- (exception, loop) => {
- warn("Failed to write %s partition entry %s: %s. Retrying." format(logType, key, exception))
- debug("Exception detail:", exception)
- if (producer != null) {
- producer.close
- }
- producer = null
- }
- )
- }
-
- private def getConsumer(): SimpleConsumer = {
- val metadataMap = TopicMetadataCache.getTopicMetadata(Set(checkpointTopic), systemName, (topics: Set[String]) => metadataStore.getTopicInfo(topics))
- val metadata = metadataMap(checkpointTopic)
- val partitionMetadata = metadata.partitionsMetadata
- .filter(_.partitionId == 0)
- .headOption
- .getOrElse(throw new KafkaCheckpointException("Tried to find partition information for partition 0 for checkpoint topic, but it didn't exist in Kafka."))
- val leader = partitionMetadata
- .leader
- .getOrElse(throw new SamzaException("No leader available for topic %s" format checkpointTopic))
-
- info("Connecting to leader %s:%d for topic %s and to fetch all checkpoint messages." format(leader.host, leader.port, checkpointTopic))
-
- new SimpleConsumer(leader.host, leader.port, socketTimeout, bufferSize, clientId)
- }
-
- private def getEarliestOffset(consumer: SimpleConsumer, topicAndPartition: TopicAndPartition): Long = consumer.earliestOrLatestOffset(topicAndPartition, OffsetRequest.EarliestTime, -1)
-
- private def getOffset(consumer: SimpleConsumer, topicAndPartition: TopicAndPartition, earliestOrLatest: Long): Long = {
- val offsetResponse = consumer.getOffsetsBefore(new OffsetRequest(Map(topicAndPartition -> PartitionOffsetRequestInfo(earliestOrLatest, 1))))
- .partitionErrorAndOffsets
- .get(topicAndPartition)
- .getOrElse(throw new KafkaCheckpointException("Unable to find offset information for %s:0" format checkpointTopic))
- // Fail or retry if there was an an issue with the offset request.
- KafkaUtil.maybeThrowException(offsetResponse.error)
-
- val offset: Long = offsetResponse
- .offsets
- .headOption
- .getOrElse(throw new KafkaCheckpointException("Got response, but no offsets defined for %s:0" format checkpointTopic))
-
- offset
- }
-
- /**
- * Read the last checkpoint for specified TaskName
- *
- * @param taskName Specific Samza taskName for which to get the last checkpoint of.
- **/
- override def readLastCheckpoint(taskName: TaskName): Checkpoint = {
- if (!taskNames.contains(taskName)) {
- throw new SamzaException(taskName + " not registered with this CheckpointManager")
- }
-
- info("Reading checkpoint for taskName " + taskName)
-
- if (taskNamesToOffsets == null) {
- info("No TaskName to checkpoint mapping provided. Reading for first time.")
- taskNamesToOffsets = readCheckpointsFromLog()
- } else {
- info("Already existing checkpoint mapping. Merging new offsets")
- taskNamesToOffsets ++= readCheckpointsFromLog()
- }
-
- val checkpoint = taskNamesToOffsets.get(taskName).getOrElse(null)
-
- info("Got checkpoint state for taskName %s: %s" format(taskName, checkpoint))
-
- checkpoint
- }
-
- /**
- * Read through entire log, discarding changelog mapping, and building map of TaskNames to Checkpoints
- */
- private def readCheckpointsFromLog(): Map[TaskName, Checkpoint] = {
- val checkpoints = mutable.Map[TaskName, Checkpoint]()
-
- def shouldHandleEntry(key: KafkaCheckpointLogKey) = key.isCheckpointKey
-
- def handleCheckpoint(payload: ByteBuffer, checkpointKey:KafkaCheckpointLogKey): Unit = {
- val taskName = checkpointKey.getCheckpointTaskName
-
- if (taskNames.contains(taskName)) {
- val checkpoint = serde.fromBytes(Utils.readBytes(payload))
-
- debug("Adding checkpoint " + checkpoint + " for taskName " + taskName)
-
- checkpoints.put(taskName, checkpoint) // replacing any existing, older checkpoints as we go
- }
- }
-
- readLog(CHECKPOINT_LOG4J_ENTRY, shouldHandleEntry, handleCheckpoint)
-
- checkpoints.toMap /* of the immutable kind */
- }
-
- /**
- * Read through entire log, discarding checkpoints, finding latest changelogPartitionMapping
- *
- * Lots of duplicated code from the checkpoint method, but will be better to refactor this code into AM-based
- * checkpoint log reading
- */
- override def readChangeLogPartitionMapping(): util.Map[TaskName, java.lang.Integer] = {
- var changelogPartitionMapping: util.Map[TaskName, java.lang.Integer] = new util.HashMap[TaskName, java.lang.Integer]()
-
- def shouldHandleEntry(key: KafkaCheckpointLogKey) = key.isChangelogPartitionMapping
-
- def handleCheckpoint(payload: ByteBuffer, checkpointKey:KafkaCheckpointLogKey): Unit = {
- changelogPartitionMapping = serde.changelogPartitionMappingFromBytes(Utils.readBytes(payload))
-
- debug("Adding changelog partition mapping" + changelogPartitionMapping)
- }
-
- readLog(CHANGELOG_PARTITION_MAPPING_LOG4j, shouldHandleEntry, handleCheckpoint)
-
- changelogPartitionMapping
- }
-
- /**
- * Common code for reading both changelog partition mapping and change log
- *
- * @param entryType What type of entry to look for within the log key's
- * @param handleEntry Code to handle an entry in the log once it's found
- */
- private def readLog(entryType:String, shouldHandleEntry: (KafkaCheckpointLogKey) => Boolean,
- handleEntry: (ByteBuffer, KafkaCheckpointLogKey) => Unit): Unit = {
- retryBackoff.run[Unit](
- loop => {
- val consumer = getConsumer()
-
- val topicAndPartition = new TopicAndPartition(checkpointTopic, 0)
-
- try {
- var offset = startingOffset.getOrElse(getEarliestOffset(consumer, topicAndPartition))
-
- info("Got offset %s for topic %s and partition 0. Attempting to fetch messages for %s." format(offset, checkpointTopic, entryType))
-
- val latestOffset = getOffset(consumer, topicAndPartition, OffsetRequest.LatestTime)
-
- info("Get latest offset %s for topic %s and partition 0." format(latestOffset, checkpointTopic))
-
- if (offset < 0) {
- info("Got offset 0 (no messages in %s) for topic %s and partition 0, so returning empty collection. If you expected the checkpoint topic to have messages, you're probably going to lose data." format (entryType, checkpointTopic))
- return
- }
-
- while (offset < latestOffset) {
- val request = new FetchRequestBuilder()
- .addFetch(checkpointTopic, 0, offset, fetchSize)
- .maxWait(500)
- .minBytes(1)
- .clientId(clientId)
- .build
-
- val fetchResponse = consumer.fetch(request)
- if (fetchResponse.hasError) {
- warn("Got error code from broker for %s: %s" format(checkpointTopic, fetchResponse.errorCode(checkpointTopic, 0)))
- val errorCode = fetchResponse.errorCode(checkpointTopic, 0)
- if (ErrorMapping.OffsetOutOfRangeCode.equals(errorCode)) {
- warn("Got an offset out of range exception while getting last entry in %s for topic %s and partition 0, so returning a null offset to the KafkaConsumer. Let it decide what to do based on its autooffset.reset setting." format (entryType, checkpointTopic))
- return
- }
- KafkaUtil.maybeThrowException(errorCode)
- }
-
- for (response <- fetchResponse.messageSet(checkpointTopic, 0)) {
- offset = response.nextOffset
- startingOffset = Some(offset) // For next time we call
-
- if (!response.message.hasKey) {
- throw new KafkaCheckpointException("Encountered message without key.")
- }
-
- val checkpointKey = KafkaCheckpointLogKey.fromBytes(Utils.readBytes(response.message.key))
-
- if (!shouldHandleEntry(checkpointKey)) {
- debug("Skipping " + entryType + " entry with key " + checkpointKey)
- } else {
- handleEntry(response.message.payload, checkpointKey)
- }
- }
- }
- } finally {
- consumer.close()
- }
-
- loop.done
- Unit
- },
-
- (exception, loop) => {
- exception match {
- case e: InvalidMessageException => throw new KafkaCheckpointException("Got InvalidMessageException from Kafka, which is unrecoverable, so fail the samza job", e)
- case e: InvalidMessageSizeException => throw new KafkaCheckpointException("Got InvalidMessageSizeException from Kafka, which is unrecoverable, so fail the samza job", e)
- case e: UnknownTopicOrPartitionException => throw new KafkaCheckpointException("Got UnknownTopicOrPartitionException from Kafka, which is unrecoverable, so fail the samza job", e)
- case e: KafkaCheckpointException => throw e
- case e: Exception =>
- warn("While trying to read last %s entry for topic %s and partition 0: %s. Retrying." format(entryType, checkpointTopic, e))
- debug("Exception detail:", e)
- }
- }
- ).getOrElse(throw new SamzaException("Failed to get entries for " + entryType + " from topic " + checkpointTopic))
-
- }
-
- def start {
- create
- validateTopic
- }
-
- def register(taskName: TaskName) {
- debug("Adding taskName " + taskName + " to " + this)
- taskNames += taskName
- }
-
- def stop = {
- if (producer != null) {
- producer.close
- }
- }
-
- def create {
- info("Attempting to create checkpoint topic %s." format checkpointTopic)
- retryBackoff.run(
- loop => {
- val zkClient = connectZk()
- try {
- AdminUtils.createTopic(
- zkClient,
- checkpointTopic,
- 1,
- replicationFactor,
- checkpointTopicProperties)
- } finally {
- zkClient.close
- }
-
- info("Created checkpoint topic %s." format checkpointTopic)
- loop.done
- },
-
- (exception, loop) => {
- exception match {
- case e: TopicExistsException =>
- info("Checkpoint topic %s already exists." format checkpointTopic)
- loop.done
- case e: Exception =>
- warn("Failed to create topic %s: %s. Retrying." format(checkpointTopic, e))
- debug("Exception detail:", e)
- }
- }
- )
- }
-
- private def validateTopic {
- info("Validating checkpoint topic %s." format checkpointTopic)
- retryBackoff.run(
- loop => {
- val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(checkpointTopic), systemName, metadataStore.getTopicInfo)
- val topicMetadata = topicMetadataMap(checkpointTopic)
- KafkaUtil.maybeThrowException(topicMetadata.errorCode)
-
- val partitionCount = topicMetadata.partitionsMetadata.length
- if (partitionCount != 1) {
- throw new KafkaCheckpointException("Checkpoint topic validation failed for topic %s because partition count %s did not match expected partition count of 1." format(checkpointTopic, topicMetadata.partitionsMetadata.length))
- }
-
- info("Successfully validated checkpoint topic %s." format checkpointTopic)
- loop.done
- },
-
- (exception, loop) => {
- exception match {
- case e: KafkaCheckpointException => throw e
- case e: Exception =>
- warn("While trying to validate topic %s: %s. Retrying." format(checkpointTopic, e))
- debug("Exception detail:", e)
- }
- }
- )
- }
-
- override def toString = "KafkaCheckpointManager [systemName=%s, checkpointTopic=%s]" format(systemName, checkpointTopic)
-}
-
-object KafkaCheckpointManager {
- val CHECKPOINT_LOG4J_ENTRY = "checkpoint log"
- val CHANGELOG_PARTITION_MAPPING_LOG4j = "changelog partition mapping"
-}
-
-/**
- * KafkaCheckpointManager handles retries, so we need two kinds of exceptions:
- * one to signal a hard failure, and the other to retry. The
- * KafkaCheckpointException is thrown to indicate a hard failure that the Kafka
- * CheckpointManager can't recover from.
- */
-class KafkaCheckpointException(s: String, t: Throwable) extends SamzaException(s, t) {
- def this(s: String) = this(s, null)
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
deleted file mode 100644
index 3dfa26a..0000000
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.checkpoint.kafka
-
-import org.apache.samza.util.Logging
-import kafka.utils.ZKStringSerializer
-import org.I0Itec.zkclient.ZkClient
-import org.apache.samza.SamzaException
-import org.apache.samza.checkpoint.CheckpointManager
-import org.apache.samza.checkpoint.CheckpointManagerFactory
-import org.apache.samza.config.Config
-import org.apache.samza.config.JobConfig.Config2Job
-import org.apache.samza.config.KafkaConfig.Config2Kafka
-import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.util.{ KafkaUtil, ClientUtilTopicMetadataStore }
-import java.util.Properties
-import scala.collection.JavaConversions._
-import org.apache.kafka.clients.producer.KafkaProducer
-
-object KafkaCheckpointManagerFactory {
- /**
- * Version number to track the format of the checkpoint log
- */
- val CHECKPOINT_LOG_VERSION_NUMBER = 1
-
- val INJECTED_PRODUCER_PROPERTIES = Map(
- "acks" -> "all",
- // Forcibly disable compression because Kafka doesn't support compression
- // on log compacted topics. Details in SAMZA-586.
- "compression.type" -> "none")
-
- // Set the checkpoint topic configs to have a very small segment size and
- // enable log compaction. This keeps job startup time small since there
- // are fewer useless (overwritten) messages to read from the checkpoint
- // topic.
- def getCheckpointTopicProperties(config: Config) = {
- val segmentBytes = config
- .getCheckpointSegmentBytes
- .getOrElse("26214400")
-
- (new Properties /: Map(
- "cleanup.policy" -> "compact",
- "segment.bytes" -> segmentBytes)) { case (props, (k, v)) => props.put(k, v); props }
- }
-}
-
-class KafkaCheckpointManagerFactory extends CheckpointManagerFactory with Logging {
- import KafkaCheckpointManagerFactory._
-
- def getCheckpointManager(config: Config, registry: MetricsRegistry): CheckpointManager = {
- val clientId = KafkaUtil.getClientId("samza-checkpoint-manager", config)
- val systemName = config
- .getCheckpointSystem
- .getOrElse(throw new SamzaException("no system defined for Kafka's checkpoint manager."))
- val producerConfig = config.getKafkaSystemProducerConfig(
- systemName,
- clientId,
- INJECTED_PRODUCER_PROPERTIES)
- val consumerConfig = config.getKafkaSystemConsumerConfig(systemName, clientId)
- val replicationFactor = config.getCheckpointReplicationFactor.getOrElse("3").toInt
- val socketTimeout = consumerConfig.socketTimeoutMs
- val bufferSize = consumerConfig.socketReceiveBufferBytes
- val fetchSize = consumerConfig.fetchMessageMaxBytes // must be > buffer size
-
- val connectProducer = () => {
- new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties)
- }
- val zkConnect = Option(consumerConfig.zkConnect)
- .getOrElse(throw new SamzaException("no zookeeper.connect defined in config"))
- val connectZk = () => {
- new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)
- }
- val jobName = config.getName.getOrElse(throw new SamzaException("Missing job name in configs"))
- val jobId = config.getJobId.getOrElse("1")
- val bootstrapServers = producerConfig.bootsrapServers
- val metadataStore = new ClientUtilTopicMetadataStore(bootstrapServers, clientId, socketTimeout)
- val checkpointTopic = getTopic(jobName, jobId)
-
- // Find out the SSPGrouperFactory class so it can be included/verified in the key
- val systemStreamPartitionGrouperFactoryString = config.getSystemStreamPartitionGrouperFactory
-
- new KafkaCheckpointManager(
- clientId,
- checkpointTopic,
- systemName,
- replicationFactor,
- socketTimeout,
- bufferSize,
- fetchSize,
- metadataStore,
- connectProducer,
- connectZk,
- systemStreamPartitionGrouperFactoryString,
- checkpointTopicProperties = getCheckpointTopicProperties(config))
- }
-
- private def getTopic(jobName: String, jobId: String) =
- "__samza_checkpoint_ver_%d_for_%s_%s" format (CHECKPOINT_LOG_VERSION_NUMBER, jobName.replaceAll("_", "-"), jobId.replaceAll("_", "-"))
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
index 02a6ea9..a1de887 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
@@ -24,7 +24,6 @@ import java.util.regex.Pattern
import org.apache.samza.util.Util
import org.apache.samza.util.Logging
-
import scala.collection.JavaConversions._
import kafka.consumer.ConsumerConfig
import java.util.{Properties, UUID}
@@ -41,10 +40,6 @@ object KafkaConfig {
val REGEX_RESOLVED_SYSTEM = "job.config.rewriter.%s.system"
val REGEX_INHERITED_CONFIG = "job.config.rewriter.%s.config"
- val CHECKPOINT_SYSTEM = "task.checkpoint.system"
- val CHECKPOINT_REPLICATION_FACTOR = "task.checkpoint.replication.factor"
- val CHECKPOINT_SEGMENT_BYTES = "task.checkpoint.segment.bytes"
-
val CHANGELOG_STREAM_REPLICATION_FACTOR = "stores.%s.changelog.replication.factor"
val CHANGELOG_STREAM_KAFKA_SETTINGS = "stores.%s.changelog.kafka."
// The default segment size to use for changelog topics
@@ -63,11 +58,6 @@ object KafkaConfig {
}
class KafkaConfig(config: Config) extends ScalaMapConfig(config) {
- // checkpoints
- def getCheckpointSystem = getOption(KafkaConfig.CHECKPOINT_SYSTEM)
- def getCheckpointReplicationFactor() = getOption(KafkaConfig.CHECKPOINT_REPLICATION_FACTOR)
- def getCheckpointSegmentBytes() = getOption(KafkaConfig.CHECKPOINT_SEGMENT_BYTES)
-
// custom consumer config
def getConsumerFetchThreshold(name: String) = getOption(KafkaConfig.CONSUMER_FETCH_THRESHOLD format name)
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/config/RegExTopicGenerator.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/RegExTopicGenerator.scala b/samza-kafka/src/main/scala/org/apache/samza/config/RegExTopicGenerator.scala
index a34c3f2..78467bf 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/config/RegExTopicGenerator.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/config/RegExTopicGenerator.scala
@@ -23,12 +23,12 @@ import org.I0Itec.zkclient.ZkClient
import kafka.utils.{ ZkUtils, ZKStringSerializer }
import org.apache.samza.config.KafkaConfig.{ Config2Kafka, REGEX_RESOLVED_STREAMS }
import org.apache.samza.SamzaException
+import org.apache.samza.util.Util
import collection.JavaConversions._
import org.apache.samza.util.Logging
import scala.collection._
import org.apache.samza.config.TaskConfig.Config2Task
import org.apache.samza.system.SystemStream
-import org.apache.samza.util.Util
import scala.util.Sorting
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
index f783c57..35086f5 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
@@ -94,6 +94,24 @@ class KafkaSystemAdmin(
brokerListString: String,
/**
+ * A method that returns a ZkClient for the Kafka system. This is invoked
+ * when the system admin is attempting to create a coordinator stream.
+ */
+ connectZk: () => ZkClient,
+
+ /**
+ * Custom properties to use when the system admin tries to create a new
+ * coordinator stream.
+ */
+ coordinatorStreamProperties: Properties = new Properties,
+
+ /**
+ * The replication factor to use when the system admin creates a new
+ * coordinator stream.
+ */
+ coordinatorStreamReplicationFactor: Int = 1,
+
+ /**
* The timeout to use for the simple consumer when fetching metadata from
* Kafka. Equivalent to Kafka's socket.timeout.ms configuration.
*/
@@ -113,12 +131,6 @@ class KafkaSystemAdmin(
clientId: String = UUID.randomUUID.toString,
/**
- * A function that returns a Zookeeper client to connect to fetch the meta
- * data information
- */
- connectZk: () => ZkClient,
-
- /**
* Replication factor for the Changelog topic in kafka
* Kafka properties to be used during the Changelog topic creation
*/
@@ -201,8 +213,39 @@ class KafkaSystemAdmin(
(exception, loop) => {
warn("Unable to fetch last offsets for streams %s due to %s. Retrying." format (streams, exception))
debug("Exception detail:", exception)
- }
- ).getOrElse(throw new SamzaException("Failed to get system stream metadata"))
+ }).getOrElse(throw new SamzaException("Failed to get system stream metadata"))
+ }
+
+ def createCoordinatorStream(streamName: String) {
+ info("Attempting to create coordinator stream %s." format streamName)
+ new ExponentialSleepStrategy(initialDelayMs = 500).run(
+ loop => {
+ val zkClient = connectZk()
+ try {
+ AdminUtils.createTopic(
+ zkClient,
+ streamName,
+ 1, // Always one partition for coordinator stream.
+ coordinatorStreamReplicationFactor,
+ coordinatorStreamProperties)
+ } finally {
+ zkClient.close
+ }
+
+ info("Created coordinator stream %s." format streamName)
+ loop.done
+ },
+
+ (exception, loop) => {
+ exception match {
+ case e: TopicExistsException =>
+ info("Coordinator stream %s already exists." format streamName)
+ loop.done
+ case e: Exception =>
+ warn("Failed to create topic %s: %s. Retrying." format (streamName, e))
+ debug("Exception detail:", e)
+ }
+ })
}
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
index c84ceb7..1629035 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
@@ -19,10 +19,13 @@
package org.apache.samza.system.kafka
+import java.util.Properties
+import org.apache.samza.SamzaException
import org.apache.samza.util.{Logging, KafkaUtil, ExponentialSleepStrategy, ClientUtilTopicMetadataStore}
import org.apache.samza.config.Config
import org.apache.samza.metrics.MetricsRegistry
import org.apache.samza.config.KafkaConfig.Config2Kafka
+import org.apache.samza.config.JobConfig.Config2Job
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.samza.system.SystemFactory
import org.apache.samza.config.StorageConfig._
@@ -98,8 +101,14 @@ class KafkaSystemFactory extends SystemFactory with Logging {
val consumerConfig = config.getKafkaSystemConsumerConfig(systemName, clientId)
val timeout = consumerConfig.socketTimeoutMs
val bufferSize = consumerConfig.socketReceiveBufferBytes
+ val zkConnect = Option(consumerConfig.zkConnect)
+ .getOrElse(throw new SamzaException("no zookeeper.connect defined in config"))
+ val connectZk = () => {
+ new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)
+ }
+ val coordinatorStreamProperties = getCoordinatorTopicProperties(config)
+ val coordinatorStreamReplicationFactor = config.getCoordinatorReplicationFactor.toInt
val storeToChangelog = config.getKafkaChangelogEnabledStores()
-
// Construct the meta information for each topic, if the replication factor is not defined, we use 2 as the number of replicas for the change log stream.
val topicMetaInformation = storeToChangelog.map{case (storeName, topicName) =>
{
@@ -112,10 +121,20 @@ class KafkaSystemFactory extends SystemFactory with Logging {
new KafkaSystemAdmin(
systemName,
bootstrapServers,
+ connectZk,
+ coordinatorStreamProperties,
+ coordinatorStreamReplicationFactor,
timeout,
bufferSize,
clientId,
- () => new ZkClient(consumerConfig.zkConnect, 6000, 6000, ZKStringSerializer),
topicMetaInformation)
}
+
+ def getCoordinatorTopicProperties(config: Config) = {
+ val segmentBytes = config.getCoordinatorSegmentBytes
+ (new Properties /: Map(
+ "cleanup.policy" -> "compact",
+ "segment.bytes" -> segmentBytes)) { case (props, (k, v)) => props.put(k, v); props }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKey.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKey.scala b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKey.scala
deleted file mode 100644
index b76d5ad..0000000
--- a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKey.scala
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.checkpoint.kafka
-
-import org.apache.samza.SamzaException
-import org.apache.samza.container.TaskName
-import org.junit.Assert._
-import org.junit.{Before, Test}
-
-class TestKafkaCheckpointLogKey {
- @Before
- def setSSPGrouperFactoryString() {
- KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString("hello")
- }
-
- @Test
- def checkpointKeySerializationRoundTrip() {
- val checkpointKey = KafkaCheckpointLogKey.getCheckpointKey(new TaskName("TN"))
- val asBytes = checkpointKey.toBytes()
- val backFromBytes = KafkaCheckpointLogKey.fromBytes(asBytes)
-
- assertEquals(checkpointKey, backFromBytes)
- assertNotSame(checkpointKey, backFromBytes)
- }
-
- @Test
- def changelogPartitionMappingKeySerializationRoundTrip() {
- val key = KafkaCheckpointLogKey.getChangelogPartitionMappingKey()
- val asBytes = key.toBytes()
- val backFromBytes = KafkaCheckpointLogKey.fromBytes(asBytes)
-
- assertEquals(key, backFromBytes)
- assertNotSame(key, backFromBytes)
- }
-
- @Test
- def differingSSPGrouperFactoriesCauseException() {
-
- val checkpointKey = KafkaCheckpointLogKey.getCheckpointKey(new TaskName("TN"))
-
- val asBytes = checkpointKey.toBytes()
-
- KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString("goodbye")
-
- var gotException = false
- try {
- KafkaCheckpointLogKey.fromBytes(asBytes)
- } catch {
- case se:SamzaException => assertEquals(new DifferingSystemStreamPartitionGrouperFactoryValues("hello", "goodbye").getMessage(), se.getCause.getMessage)
- gotException = true
- }
-
- assertTrue("Should have had an exception since ssp grouper factories didn't match", gotException)
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
deleted file mode 100644
index 7d4bea8..0000000
--- a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
+++ /dev/null
@@ -1,211 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.checkpoint.kafka
-
-import kafka.admin.AdminUtils
-import kafka.common.InvalidMessageSizeException
-import kafka.common.UnknownTopicOrPartitionException
-import kafka.message.InvalidMessageException
-
-import kafka.server.KafkaConfig
-import kafka.server.KafkaServer
-import kafka.utils.TestUtils
-import kafka.utils.TestZKUtils
-import kafka.utils.Utils
-import kafka.utils.ZKStringSerializer
-import kafka.zk.EmbeddedZookeeper
-
-import org.I0Itec.zkclient.ZkClient
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.config.{KafkaProducerConfig, MapConfig}
-import org.apache.samza.container.TaskName
-import org.apache.samza.container.grouper.stream.GroupByPartitionFactory
-import org.apache.samza.serializers.CheckpointSerde
-import org.apache.samza.system.SystemStreamPartition
-import org.apache.samza.util.{ ClientUtilTopicMetadataStore, TopicMetadataStore }
-import org.apache.samza.{ SamzaException, Partition }
-import org.junit.Assert._
-import org.junit.{ AfterClass, BeforeClass, Test }
-
-import scala.collection._
-import scala.collection.JavaConversions._
-import org.apache.kafka.clients.producer.{ProducerConfig, KafkaProducer}
-
-object TestKafkaCheckpointManager {
- val checkpointTopic = "checkpoint-topic"
- val serdeCheckpointTopic = "checkpoint-topic-invalid-serde"
- val zkConnect: String = TestZKUtils.zookeeperConnect
- var zkClient: ZkClient = null
- val zkConnectionTimeout = 6000
- val zkSessionTimeout = 6000
-
- val brokerId1 = 0
- val brokerId2 = 1
- val brokerId3 = 2
- val ports = TestUtils.choosePorts(3)
- val (port1, port2, port3) = (ports(0), ports(1), ports(2))
-
- val props1 = TestUtils.createBrokerConfig(brokerId1, port1)
- props1.put("controlled.shutdown.enable", "true")
- val props2 = TestUtils.createBrokerConfig(brokerId2, port2)
- props1.put("controlled.shutdown.enable", "true")
- val props3 = TestUtils.createBrokerConfig(brokerId3, port3)
- props1.put("controlled.shutdown.enable", "true")
-
- val config = new java.util.HashMap[String, Object]()
- val brokers = "localhost:%d,localhost:%d,localhost:%d" format (port1, port2, port3)
- config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
- config.put("acks", "all")
- config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, new Integer(1))
- config.put(ProducerConfig.RETRIES_CONFIG, new Integer(java.lang.Integer.MAX_VALUE-1))
- config.putAll(KafkaCheckpointManagerFactory.INJECTED_PRODUCER_PROPERTIES)
- val producerConfig = new KafkaProducerConfig("kafka", "i001", config)
- val partition = new Partition(0)
- val cp1 = new Checkpoint(Map(new SystemStreamPartition("kafka", "topic", partition) -> "123"))
- val cp2 = new Checkpoint(Map(new SystemStreamPartition("kafka", "topic", partition) -> "12345"))
- var zookeeper: EmbeddedZookeeper = null
- var server1: KafkaServer = null
- var server2: KafkaServer = null
- var server3: KafkaServer = null
- var metadataStore: TopicMetadataStore = null
-
- val systemStreamPartitionGrouperFactoryString = classOf[GroupByPartitionFactory].getCanonicalName
-
- @BeforeClass
- def beforeSetupServers {
- zookeeper = new EmbeddedZookeeper(zkConnect)
- server1 = TestUtils.createServer(new KafkaConfig(props1))
- server2 = TestUtils.createServer(new KafkaConfig(props2))
- server3 = TestUtils.createServer(new KafkaConfig(props3))
- metadataStore = new ClientUtilTopicMetadataStore(brokers, "some-job-name")
- }
-
- @AfterClass
- def afterCleanLogDirs {
- server1.shutdown
- server1.awaitShutdown()
- server2.shutdown
- server2.awaitShutdown()
- server3.shutdown
- server3.awaitShutdown()
- Utils.rm(server1.config.logDirs)
- Utils.rm(server2.config.logDirs)
- Utils.rm(server3.config.logDirs)
- zookeeper.shutdown
- }
-}
-
-class TestKafkaCheckpointManager {
- import TestKafkaCheckpointManager._
-
- @Test
- def testCheckpointShouldBeNullIfCheckpointTopicDoesNotExistShouldBeCreatedOnWriteAndShouldBeReadableAfterWrite {
- val kcm = getKafkaCheckpointManager
- val taskName = new TaskName(partition.toString)
- kcm.register(taskName)
- kcm.start
- // check that log compaction is enabled.
- val zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)
- val topicConfig = AdminUtils.fetchTopicConfig(zkClient, checkpointTopic)
- zkClient.close
- assertEquals("compact", topicConfig.get("cleanup.policy"))
- assertEquals("26214400", topicConfig.get("segment.bytes"))
- // read before topic exists should result in a null checkpoint
- var readCp = kcm.readLastCheckpoint(taskName)
- assertNull(readCp)
- // create topic the first time around
- kcm.writeCheckpoint(taskName, cp1)
- readCp = kcm.readLastCheckpoint(taskName)
- assertEquals(cp1, readCp)
- // should get an exception if partition doesn't exist
- try {
- readCp = kcm.readLastCheckpoint(new TaskName(new Partition(1).toString))
- fail("Expected a SamzaException, since only one partition (partition 0) should exist.")
- } catch {
- case e: SamzaException => None // expected
- case _: Exception => fail("Expected a SamzaException, since only one partition (partition 0) should exist.")
- }
- // writing a second message should work, too
- kcm.writeCheckpoint(taskName, cp2)
- readCp = kcm.readLastCheckpoint(taskName)
- assertEquals(cp2, readCp)
- kcm.stop
- }
-
- @Test
- def testUnrecoverableKafkaErrorShouldThrowKafkaCheckpointManagerException {
- val exceptions = List("InvalidMessageException", "InvalidMessageSizeException", "UnknownTopicOrPartitionException")
- exceptions.foreach { exceptionName =>
- val kcm = getKafkaCheckpointManagerWithInvalidSerde(exceptionName)
- val taskName = new TaskName(partition.toString)
- kcm.register(taskName)
- kcm.start
- kcm.writeCheckpoint(taskName, cp1)
- // because serde will throw unrecoverable errors, it should result a KafkaCheckpointException
- try {
- kcm.readLastCheckpoint(taskName)
- fail("Expected a KafkaCheckpointException.")
- } catch {
- case e: KafkaCheckpointException => None
- }
- kcm.stop
- }
- }
-
- private def getKafkaCheckpointManager = new KafkaCheckpointManager(
- clientId = "some-client-id",
- checkpointTopic = checkpointTopic,
- systemName = "kafka",
- replicationFactor = 3,
- socketTimeout = 30000,
- bufferSize = 64 * 1024,
- fetchSize = 300 * 1024,
- metadataStore = metadataStore,
- connectProducer = () => new KafkaProducer(producerConfig.getProducerProperties),
- connectZk = () => new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer),
- systemStreamPartitionGrouperFactoryString = systemStreamPartitionGrouperFactoryString,
- checkpointTopicProperties = KafkaCheckpointManagerFactory.getCheckpointTopicProperties(new MapConfig(Map[String, String]())))
-
- // inject serde. Kafka exceptions will be thrown when serde.fromBytes is called
- private def getKafkaCheckpointManagerWithInvalidSerde(exception: String) = new KafkaCheckpointManager(
- clientId = "some-client-id-invalid-serde",
- checkpointTopic = serdeCheckpointTopic,
- systemName = "kafka",
- replicationFactor = 3,
- socketTimeout = 30000,
- bufferSize = 64 * 1024,
- fetchSize = 300 * 1024,
- metadataStore = metadataStore,
- connectProducer = () => new KafkaProducer(producerConfig.getProducerProperties),
- connectZk = () => new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer),
- systemStreamPartitionGrouperFactoryString = systemStreamPartitionGrouperFactoryString,
- serde = new InvalideSerde(exception),
- checkpointTopicProperties = KafkaCheckpointManagerFactory.getCheckpointTopicProperties(new MapConfig(Map[String, String]())))
-
- class InvalideSerde(exception: String) extends CheckpointSerde {
- override def fromBytes(bytes: Array[Byte]): Checkpoint = {
- exception match {
- case "InvalidMessageException" => throw new InvalidMessageException
- case "InvalidMessageSizeException" => throw new InvalidMessageSizeException
- case "UnknownTopicOrPartitionException" => throw new UnknownTopicOrPartitionException
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
index 0380d35..d260f2d 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
@@ -100,15 +100,15 @@ object TestKafkaSystemAdmin {
REPLICATION_FACTOR)
}
- def validateTopic(expectedPartitionCount: Int) {
+ def validateTopic(topic: String, expectedPartitionCount: Int) {
var done = false
var retries = 0
val maxRetries = 100
while (!done && retries < maxRetries) {
try {
- val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(TOPIC), "kafka", metadataStore.getTopicInfo)
- val topicMetadata = topicMetadataMap(TOPIC)
+ val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(topic), "kafka", metadataStore.getTopicInfo)
+ val topicMetadata = topicMetadataMap(topic)
val errorCode = topicMetadata.errorCode
KafkaUtil.maybeThrowException(errorCode)
@@ -207,10 +207,9 @@ class TestKafkaSystemAdmin {
@Test
def testShouldGetOldestNewestAndNextOffsets {
-
// Create an empty topic with 50 partitions, but with no offsets.
createTopic
- validateTopic(50)
+ validateTopic(TOPIC, 50)
// Verify the empty topic behaves as expected.
var metadata = systemAdmin.getSystemStreamMetadata(Set(TOPIC))
@@ -271,7 +270,6 @@ class TestKafkaSystemAdmin {
@Test
def testNonExistentTopic {
-
val initialOffsets = systemAdmin.getSystemStreamMetadata(Set("non-existent-topic"))
val metadata = initialOffsets.getOrElse("non-existent-topic", fail("missing metadata"))
assertEquals(metadata, new SystemStreamMetadata("non-existent-topic", Map(
@@ -289,7 +287,21 @@ class TestKafkaSystemAdmin {
assertEquals("3", offsetsAfter(ssp2))
}
- class KafkaSystemAdminWithTopicMetadataError extends KafkaSystemAdmin("test", brokers, connectZk = () => new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)) {
+ @Test
+ def testShouldCreateCoordinatorStream {
+ val topic = "test-coordinator-stream"
+ val systemAdmin = new KafkaSystemAdmin("test", brokers, () => new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer), coordinatorStreamReplicationFactor = 3)
+ systemAdmin.createCoordinatorStream(topic)
+ validateTopic(topic, 1)
+ val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(topic), "kafka", metadataStore.getTopicInfo)
+ assertTrue(topicMetadataMap.contains(topic))
+ val topicMetadata = topicMetadataMap(topic)
+ val partitionMetadata = topicMetadata.partitionsMetadata.head
+ assertEquals(0, partitionMetadata.partitionId)
+ assertEquals(3, partitionMetadata.replicas.size)
+ }
+
+ class KafkaSystemAdminWithTopicMetadataError extends KafkaSystemAdmin("test", brokers, () => new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)) {
import kafka.api.{ TopicMetadata, TopicMetadataResponse }
// Simulate Kafka telling us that the leader for the topic is not available
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
index d3f25c0..d3616fe 100644
--- a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
+++ b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
@@ -28,9 +28,13 @@ import org.apache.log4j.Logger;
import org.apache.log4j.spi.LoggingEvent;
import org.apache.samza.SamzaException;
import org.apache.samza.config.Config;
+import org.apache.samza.config.MapConfig;
import org.apache.samza.config.Log4jSystemConfig;
import org.apache.samza.config.SerializerConfig;
import org.apache.samza.config.ShellCommandConfig;
+import org.apache.samza.config.StreamConfig;
+import org.apache.samza.config.SystemConfig;
+import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.job.model.JobModel;
import org.apache.samza.logging.log4j.serializers.LoggingEventJsonSerdeFactory;
import org.apache.samza.metrics.MetricsRegistryMap;
@@ -177,7 +181,8 @@ public class StreamAppender extends AppenderSkeleton {
try {
if (isApplicationMaster) {
- config = SamzaObjectMapper.getObjectMapper().readValue(System.getenv(ShellCommandConfig.ENV_CONFIG()), Config.class);
+ Config coordinatorSystemConfig = new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG()), Config.class));
+ config = JobCoordinator.apply(coordinatorSystemConfig).jobModel().getConfig();
} else {
String url = System.getenv(ShellCommandConfig.ENV_COORDINATOR_URL());
config = SamzaObjectMapper.getObjectMapper().readValue(Util.read(new URL(url), 30000), JobModel.class).getConfig();
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-test/src/main/config/join/common.properties
----------------------------------------------------------------------
diff --git a/samza-test/src/main/config/join/common.properties b/samza-test/src/main/config/join/common.properties
index ad10aac..ac87e81 100644
--- a/samza-test/src/main/config/join/common.properties
+++ b/samza-test/src/main/config/join/common.properties
@@ -23,7 +23,7 @@ yarn.package.path=<YARN.PACKAGE.PATH>
job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
task.checkpoint.factory=org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory
-task.checkpoint.system=kafka-checkpoints
+task.checkpoint.system=kafka
task.checkpoint.replication.factor=1
serializers.registry.string.class=org.apache.samza.serializers.StringSerdeFactory
@@ -36,11 +36,5 @@ systems.kafka.producer.bootstrap.servers=localhost:9092
systems.kafka.samza.key.serde=string
systems.kafka.samza.msg.serde=string
- # Checkpoints System
-systems.kafka-checkpoints.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory
-systems.kafka-checkpoints.producer.bootstrap.servers=localhost:9092
-systems.kafka-checkpoints.consumer.zookeeper.connect=localhost:2181
-
yarn.container.retry.count=-1
yarn.container.retry.window.ms=60000
-
http://git-wip-us.apache.org/repos/asf/samza/blob/23fb2e1c/samza-test/src/main/config/negate-number.properties
----------------------------------------------------------------------
diff --git a/samza-test/src/main/config/negate-number.properties b/samza-test/src/main/config/negate-number.properties
index b9f898c..26fa1e3 100644
--- a/samza-test/src/main/config/negate-number.properties
+++ b/samza-test/src/main/config/negate-number.properties
@@ -38,3 +38,6 @@ systems.kafka.samza.key.serde=string
systems.kafka.samza.offset.default=oldest
systems.kafka.consumer.zookeeper.connect=localhost:2181/
systems.kafka.producer.bootstrap.servers=localhost:9092
+
+#Coordinator replication factor
+job.coordinator.replication.factor=1