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/11 00:07:01 UTC
git commit: SAMZA-141;
make task instance use systemstream instead of systemstream partition
in process method. this fixes a serious checkpointing issue that causes a
container to restart at its prior starting point every time a container
restarts.
Updated Branches:
refs/heads/master 830a23e17 -> bf1904d64
SAMZA-141; make task instance use systemstream instead of systemstream partition in process method. this fixes a serious checkpointing issue that causes a container to restart at its prior starting point every time a container restarts.
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/bf1904d6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/bf1904d6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/bf1904d6
Branch: refs/heads/master
Commit: bf1904d64389412fd20d942c625741c621cb4938
Parents: 830a23e
Author: Chris Riccomini <cr...@criccomi-ld.linkedin.biz>
Authored: Mon Feb 10 14:51:08 2014 -0800
Committer: Chris Riccomini <cr...@criccomi-ld.linkedin.biz>
Committed: Mon Feb 10 14:51:08 2014 -0800
----------------------------------------------------------------------
.../apache/samza/container/TaskInstance.scala | 2 +-
.../samza/serializers/CheckpointSerde.scala | 11 ++-
.../samza/container/TestTaskInstance.scala | 72 ++++++++++++++++++++
.../samza/serializers/TestCheckpointSerde.scala | 58 ++++++++++++++++
4 files changed, 141 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/bf1904d6/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 b30dd27..974db63 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
@@ -171,7 +171,7 @@ class TaskInstance(
trace("Updating offset map for partition: %s, %s, %s" format (partition, envelope.getSystemStreamPartition, envelope.getOffset))
- offsets += envelope.getSystemStreamPartition -> envelope.getOffset
+ offsets += envelope.getSystemStreamPartition.getSystemStream -> envelope.getOffset
}
def window(coordinator: ReadableCoordinator) {
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/bf1904d6/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
index 574c584..0d82183 100644
--- a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
@@ -23,6 +23,7 @@ import scala.collection.JavaConversions._
import org.codehaus.jackson.map.ObjectMapper
import org.apache.samza.system.SystemStream
import org.apache.samza.checkpoint.Checkpoint
+import org.apache.samza.SamzaException
class CheckpointSerde extends Serde[Checkpoint] {
val jsonMapper = new ObjectMapper()
@@ -56,7 +57,15 @@ class CheckpointSerde extends Serde[Checkpoint] {
.groupBy(_._2)
// There should only ever be one SystemStream to offset mapping, so just
// grab the first element from the tuple list for each stream.
- .map { case (streamName, tuples) => (streamName, tuples.head._3) }
+ .map {
+ case (streamName, tuples) => {
+ // If there's more than one offset, something is seriously wrong.
+ if (tuples.size != 1) {
+ throw new SamzaException("Got %s offsets for %s. Expected only one offset, so failing." format (tuples.size, streamName))
+ }
+ (streamName, tuples.head._3)
+ }
+ }
.toMap)
(systemName, streamToOffestMap)
}.toMap)
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/bf1904d6/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
new file mode 100644
index 0000000..5e9dd07
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
@@ -0,0 +1,72 @@
+/*
+ * 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.container
+
+import org.junit.Assert._
+import org.junit.Test
+import org.apache.samza.system.IncomingMessageEnvelope
+import org.apache.samza.system.SystemProducers
+import org.apache.samza.task.MessageCollector
+import org.apache.samza.task.StreamTask
+import org.apache.samza.system.SystemConsumers
+import org.apache.samza.task.TaskCoordinator
+import org.apache.samza.config.MapConfig
+import org.apache.samza.Partition
+import org.apache.samza.system.chooser.RoundRobinChooser
+import org.apache.samza.system.SystemProducer
+import org.apache.samza.serializers.SerdeManager
+import org.apache.samza.system.SystemConsumer
+import org.apache.samza.system.SystemStream
+import org.apache.samza.system.SystemStreamPartition
+import org.apache.samza.task.ReadableCoordinator
+
+class TestTaskInstance {
+ @Test
+ def testOffsetsAreUpdatedOnProcess {
+ val task = new StreamTask {
+ def process(envelope: IncomingMessageEnvelope, collector: MessageCollector, coordinator: TaskCoordinator) {
+ }
+ }
+ val config = new MapConfig
+ val partition = new Partition(0)
+ val containerName = "test-container"
+ val consumerMultiplexer = new SystemConsumers(
+ new RoundRobinChooser,
+ Map[String, SystemConsumer]())
+ val producerMultiplexer = new SystemProducers(
+ Map[String, SystemProducer](),
+ new SerdeManager)
+ val taskInstance: TaskInstance = new TaskInstance(
+ task,
+ partition,
+ config,
+ new TaskInstanceMetrics,
+ consumerMultiplexer: SystemConsumers,
+ producerMultiplexer: SystemProducers)
+ val systemStream = new SystemStream("test-system", "test-stream")
+ // Pretend our last checkpointed offset was 1.
+ taskInstance.offsets += systemStream -> "1"
+ // Pretend we got a message with offset 2.
+ taskInstance.process(new IncomingMessageEnvelope(new SystemStreamPartition("test-system", "test-stream", new Partition(0)), "2", null, null), new ReadableCoordinator)
+ // Check to see if the offset map has been properly updated with offset 2.
+ assertEquals(1, taskInstance.offsets.size)
+ assertEquals("2", taskInstance.offsets(systemStream))
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/bf1904d6/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
new file mode 100644
index 0000000..70d8c80
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
@@ -0,0 +1,58 @@
+/*
+ * 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.serializers
+
+import org.junit.Assert._
+import org.junit.Test
+import org.apache.samza.system.SystemStream
+import org.apache.samza.checkpoint.Checkpoint
+import scala.collection.JavaConversions._
+import org.apache.samza.system.SystemStreamPartition
+import org.apache.samza.SamzaException
+import org.apache.samza.Partition
+
+class TestCheckpointSerde {
+ @Test
+ def testExactlyOneOffset {
+ val serde = new CheckpointSerde
+ var offsets = Map[SystemStream, String]()
+ val systemStream = new SystemStream("test-system", "test-stream")
+ offsets += systemStream -> "1"
+ val deserializedOffsets = serde.fromBytes(serde.toBytes(new Checkpoint(offsets)))
+ assertEquals("1", deserializedOffsets.getOffsets.get(systemStream))
+ assertEquals(1, deserializedOffsets.getOffsets.size)
+ }
+
+ @Test
+ def testMoreThanOneOffsetShouldFail {
+ val serde = new CheckpointSerde
+ var offsets = Map[SystemStream, String]()
+ // Since SS != SSP with same system and stream name, this should result in
+ // two offsets for one system stream in the serde.
+ offsets += new SystemStream("test-system", "test-stream") -> "1"
+ offsets += new SystemStreamPartition("test-system", "test-stream", new Partition(0)) -> "2"
+ try {
+ serde.toBytes(new Checkpoint(offsets))
+ fail("Expected to fail with more than one offset for a single SystemStream.")
+ } catch {
+ case e: SamzaException => // expected this
+ }
+ }
+}
\ No newline at end of file