You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by cr...@apache.org on 2014/02/28 18:27:03 UTC
git commit: SAMZA-142;
changelog stores should restore from beginning of stream, not the end
Repository: incubator-samza
Updated Branches:
refs/heads/master 534e9f548 -> f50f022c7
SAMZA-142; changelog stores should restore from beginning of stream, not the end
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/f50f022c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/f50f022c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/f50f022c
Branch: refs/heads/master
Commit: f50f022c7d0fbe648412c26c9d6dc677e7758006
Parents: 534e9f5
Author: Chris Riccomini <cr...@criccomi-mn.linkedin.biz>
Authored: Fri Feb 28 09:26:54 2014 -0800
Committer: Chris Riccomini <cr...@criccomi-mn.linkedin.biz>
Committed: Fri Feb 28 09:26:54 2014 -0800
----------------------------------------------------------------------
.../samza/system/SystemStreamMetadata.java | 8 +++--
.../apache/samza/container/SamzaContainer.scala | 31 ++++++++++++++----
.../apache/samza/container/TaskInstance.scala | 2 +-
.../samza/storage/TaskStorageManager.scala | 24 ++++++++++----
.../samza/container/TestSamzaContainer.scala | 2 +-
.../test/integration/TestStatefulTask.scala | 34 ++++----------------
6 files changed, 55 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/samza-api/src/main/java/org/apache/samza/system/SystemStreamMetadata.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/system/SystemStreamMetadata.java b/samza-api/src/main/java/org/apache/samza/system/SystemStreamMetadata.java
index 48aa224..32e142a 100644
--- a/samza-api/src/main/java/org/apache/samza/system/SystemStreamMetadata.java
+++ b/samza-api/src/main/java/org/apache/samza/system/SystemStreamMetadata.java
@@ -108,7 +108,8 @@ public class SystemStreamMetadata {
* partition given. If a partition has two messages with offsets 0
* and 1, respectively, then this method would return 0 for the
* oldest offset. This offset is useful when one wishes to read all
- * messages in a stream from the very beginning.
+ * messages in a stream from the very beginning. A null value means
+ * the stream is empty.
*/
public String getOldestOffset() {
return oldestOffset;
@@ -120,7 +121,7 @@ public class SystemStreamMetadata {
* respectively, then this method would return 1 for the newest
* offset. This offset is useful when one wishes to see if all
* messages have been read from a stream (offset of last message
- * read == newest offset).
+ * read == newest offset). A null value means the stream is empty.
*/
public String getNewestOffset() {
return newestOffset;
@@ -131,7 +132,8 @@ public class SystemStreamMetadata {
* stream for the partition given. If a partition has two messages
* with offsets 0 and 1, respectively, then this method would return
* 2 for the upcoming offset. This offset is useful when one wishes
- * to pick up reading at the very end of a stream.
+ * to pick up reading at the very end of a stream. A null value
+ * means the stream is empty.
*/
public String getUpcomingOffset() {
return upcomingOffset;
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index ebe3ed0..9c23244 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -127,7 +127,7 @@ object SamzaContainer extends Logging {
info("Got system factories: %s" format systemFactories.keys)
- val inputStreamMetadata = getInputStreamMetadata(inputStreams, systemAdmins)
+ val inputStreamMetadata = getStreamMetadata(inputStreams.map(_.getSystemStream), systemAdmins)
info("Got input stream metadata: %s" format inputStreamMetadata)
@@ -229,6 +229,10 @@ object SamzaContainer extends Logging {
info("Got change log system streams: %s" format changeLogSystemStreams)
+ val changeLogMetadata = getStreamMetadata(changeLogSystemStreams.values.toSet, systemAdmins)
+
+ info("Got change log stream metadata: %s" format changeLogMetadata)
+
val serdeManager = new SerdeManager(
serdes = serdes,
systemKeySerdes = systemKeySerdes,
@@ -393,11 +397,16 @@ object SamzaContainer extends Logging {
info("Got task stores: %s" format taskStores)
+ val changeLogOldestOffsets = getChangeLogOldestOffsetsForPartition(partition, changeLogMetadata)
+
+ info("Assigning oldest change log offsets for partition %s: %s" format (partition, changeLogOldestOffsets))
+
val storageManager = new TaskStorageManager(
partition = partition,
taskStores = taskStores,
storeConsumers = storeConsumers,
changeLogSystemStreams = changeLogSystemStreams,
+ changeLogOldestOffsets = changeLogOldestOffsets,
storeBaseDir = storeBaseDir)
val inputStreamsForThisPartition = inputStreams.filter(_.getPartition.equals(partition)).map(_.getSystemStream)
@@ -438,13 +447,11 @@ object SamzaContainer extends Logging {
}
/**
- * Builds a map from SystemStream to SystemStreamMetadata for all input
- * streams using SystemAdmins to fetch metadata.
+ * Builds a map from SystemStream to SystemStreamMetadata for all streams
+ * using SystemAdmins to fetch metadata.
*/
- def getInputStreamMetadata(inputStreams: Set[SystemStreamPartition], systemAdmins: Map[String, SystemAdmin]): Map[SystemStream, SystemStreamMetadata] = {
- inputStreams
- .map(_.getSystemStream)
- .toSet
+ def getStreamMetadata(streams: Set[SystemStream], systemAdmins: Map[String, SystemAdmin]): Map[SystemStream, SystemStreamMetadata] = {
+ streams
.groupBy[String](_.getSystem)
.flatMap {
case (systemName, systemStreams) =>
@@ -457,6 +464,16 @@ object SamzaContainer extends Logging {
}
.toMap
}
+
+ /**
+ * Builds a map from SystemStreamPartition to oldest offset for changelogs.
+ */
+ def getChangeLogOldestOffsetsForPartition(partition: Partition, inputStreamMetadata: Map[SystemStream, SystemStreamMetadata]): Map[SystemStream, String] = {
+ inputStreamMetadata
+ .mapValues(_.getSystemStreamPartitionMetadata.get(partition))
+ .filter(_._2 != null)
+ .mapValues(_.getOldestOffset)
+ }
}
class SamzaContainer(
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index b42031e..17449ee 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -100,7 +100,7 @@ class TaskInstance(
if (storageManager != null) {
debug("Starting storage manager for partition: %s." format partition)
- storageManager.init(collector)
+ storageManager.init
} else {
debug("Skipping storage manager initialization for partition: %s." format partition)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala b/samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala
index f4c0194..7214151 100644
--- a/samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala
+++ b/samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala
@@ -27,8 +27,8 @@ import org.apache.samza.system.SystemConsumer
import org.apache.samza.system.SystemStream
import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.system.SystemStreamPartitionIterator
-import org.apache.samza.task.MessageCollector
import org.apache.samza.util.Util
+import org.apache.samza.SamzaException
object TaskStorageManager {
def getStoreDir(storeBaseDir: File, storeName: String) = {
@@ -48,14 +48,17 @@ class TaskStorageManager(
taskStores: Map[String, StorageEngine] = Map(),
storeConsumers: Map[String, SystemConsumer] = Map(),
changeLogSystemStreams: Map[String, SystemStream] = Map(),
+ changeLogOldestOffsets: Map[SystemStream, String] = Map(),
storeBaseDir: File = new File(System.getProperty("user.dir"), "state")) extends Logging {
+ var taskStoresToRestore = taskStores
+
def apply(storageEngineName: String) = taskStores(storageEngineName)
- def init(collector: MessageCollector) {
+ def init {
cleanBaseDirs
startConsumers
- restoreStores(collector)
+ restoreStores
stopConsumers
}
@@ -78,19 +81,26 @@ class TaskStorageManager(
for ((storeName, systemStream) <- changeLogSystemStreams) {
val systemStreamPartition = new SystemStreamPartition(systemStream, partition)
val consumer = storeConsumers(storeName)
+ val offset = changeLogOldestOffsets.getOrElse(systemStream, throw new SamzaException("Missing a change log offset for %s." format systemStreamPartition))
+
+ if (offset != null) {
+ info("Registering change log consumer with offset %s for %s." format (offset, systemStreamPartition))
- debug("Registering consumer for system stream partition %s." format systemStreamPartition)
+ consumer.register(systemStreamPartition, offset)
+ } else {
+ info("Skipping change log restoration for %s because stream appears to be empty (offset was null)." format systemStreamPartition)
- consumer.register(systemStreamPartition, null)
+ taskStoresToRestore -= storeName
+ }
}
storeConsumers.values.foreach(_.start)
}
- private def restoreStores(collector: MessageCollector) {
+ private def restoreStores {
debug("Restoring stores.")
- for ((storeName, store) <- taskStores) {
+ for ((storeName, store) <- taskStoresToRestore) {
if (changeLogSystemStreams.contains(storeName)) {
val systemStream = changeLogSystemStreams(storeName)
val systemStreamPartition = new SystemStreamPartition(systemStream, partition)
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/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 555c95c..a2d5820 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
@@ -52,7 +52,7 @@ class TestSamzaContainer {
new SystemStreamPartition("test", "stream2", new Partition(0)),
new SystemStreamPartition("test", "stream2", new Partition(1)))
val systemAdmins = Map("test" -> new SinglePartitionWithoutOffsetsSystemAdmin)
- val metadata = SamzaContainer.getInputStreamMetadata(inputStreams, systemAdmins)
+ val metadata = SamzaContainer.getStreamMetadata(inputStreams.map(_.getSystemStream).toSet, systemAdmins)
assertNotNull(metadata)
assertEquals(2, metadata.size)
val stream1Metadata = metadata(new SystemStream("test", "stream1"))
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f50f022c/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
index 387cf44..d36de26 100644
--- a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
+++ b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
@@ -214,7 +214,7 @@ class TestStatefulTask {
"stores.mystore.changelog" -> "kafka.mystore",
"systems.kafka.samza.factory" -> "org.apache.samza.system.kafka.KafkaSystemFactory",
"systems.kafka.consumer.zookeeper.connect" -> zkConnect,
- "systems.kafka.consumer.auto.offset.reset" -> "smallest",
+ "systems.kafka.consumer.auto.offset.reset" -> "largest",
"systems.kafka.producer.metadata.broker.list" -> ("localhost:%s" format port1),
"systems.kafka.samza.msg.serde" -> "string")
@@ -254,50 +254,30 @@ class TestStatefulTask {
def testShouldRestoreStore {
val (job, task) = startJob
- // Validate that restored is empty.
+ // Validate that restored has expected data.
assertEquals(3, task.restored.size)
assertTrue(task.restored.contains("1"))
assertTrue(task.restored.contains("2"))
assertTrue(task.restored.contains("3"))
- var count = 0
-
- // We should get the original four messages in the stream (1,2,3,2).
- // Note that this will trigger four new outgoing messages to the STATE_TOPIC.
- while (task.received.size < 4 && count < 100) {
- Thread.sleep(600)
- count += 1
- }
-
- assertTrue(count < 100)
-
- // Reset the count down latch after the 4 messages come in.
- task.awaitMessage
-
// Send some messages to input stream.
send(task, "4")
send(task, "5")
send(task, "5")
// Validate that messages appear in store stream.
- val messages = readAll(STATE_TOPIC, 10, "testShouldRestoreStore")
+ val messages = readAll(STATE_TOPIC, 6, "testShouldRestoreStore")
- assertEquals(11, messages.length)
+ assertEquals(7, messages.length)
// From initial start.
assertEquals("1", messages(0))
assertEquals("2", messages(1))
assertEquals("3", messages(2))
assertEquals("2", messages(3))
- // From second start, since we're we have no checkpoint manager defined,
- // and are starting back at offset 0 for input stream, again.
- assertEquals("1", messages(4))
- assertEquals("2", messages(5))
- assertEquals("3", messages(6))
- assertEquals("2", messages(7))
// From sending in this method.
- assertEquals("4", messages(8))
- assertEquals("5", messages(9))
- assertEquals("5", messages(10))
+ assertEquals("4", messages(4))
+ assertEquals("5", messages(5))
+ assertEquals("5", messages(6))
stopJob(job)
}