You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by pm...@apache.org on 2019/01/23 00:20:32 UTC
samza git commit: SAMZA-2042: Add support for inspecting/modifying
checkpoints before they're written.
Repository: samza
Updated Branches:
refs/heads/master 1cac1c8e5 -> 18ebfce33
SAMZA-2042: Add support for inspecting/modifying checkpoints before they're written.
Author: Prateek Maheshwari <pm...@apache.org>
Reviewers: Jagadish Venkatraman <vj...@gmail.com>, Shanthoosh Venkataraman <sv...@linkedin.com>
Closes #859 from prateekm/checkpoint-listener
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/18ebfce3
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/18ebfce3
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/18ebfce3
Branch: refs/heads/master
Commit: 18ebfce338ec3a51554a4303566ec706614de68a
Parents: 1cac1c8
Author: Prateek Maheshwari <pm...@apache.org>
Authored: Tue Jan 22 16:20:29 2019 -0800
Committer: Prateek Maheshwari <pm...@apache.org>
Committed: Tue Jan 22 16:20:29 2019 -0800
----------------------------------------------------------------------
.../samza/checkpoint/CheckpointListener.java | 38 +++++++++--
.../apache/samza/checkpoint/OffsetManager.scala | 62 ++++++++++++++---
.../samza/checkpoint/TestOffsetManager.scala | 72 ++++++++++++++++++++
3 files changed, 157 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/18ebfce3/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java
index f81167a..60df3c4 100644
--- a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java
@@ -22,11 +22,39 @@ import java.util.Map;
import org.apache.samza.system.SystemStreamPartition;
/**
- * This interface should be implemented by SystemConsumers which want to be notified about checkpoint values
- * of the most recent checkpoint.
- * Note: For broadcast streams, different tasks may register the same ssp with different values. It
- * is the callee's responsibility to select the lowest one.
+ * SystemConsumers that want to be notified about checkpoints for their SSPs and potentially modify them may
+ * implement this interface.
+ *
+ * Note: For broadcast streams, different Tasks may checkpoint the same ssp with different offsets. It is the
+ * SystemConsumer's responsibility to select the appropriate (e.g., lowest) offset for its own use.
*/
public interface CheckpointListener {
- void onCheckpoint(Map<SystemStreamPartition, String> offsets);
+ /**
+ * Called before writing the checkpoint for each Task in the Container processing SSPs on this system.
+ * Implementations may modify the Task offsets to be checkpointed, or return the provided offsets as-is.
+ *
+ * @param offsets offsets about to be checkpointed.
+ * @return potentially modified offsets to be checkpointed.
+ */
+ default Map<SystemStreamPartition, String> beforeCheckpoint(Map<SystemStreamPartition, String> offsets) {
+ return offsets;
+ }
+
+ /**
+ * Called after writing the checkpoint for each Task in the Container processing SSPs on this system.
+ *
+ * @param offsets offsets that were checkpointed.
+ */
+ default void afterCheckpoint(Map<SystemStreamPartition, String> offsets) { }
+
+ /**
+ * Called after writing the checkpoint for each Task in the Container processing SSPs on this system.
+ * This method is deprecated. Implement {@link #afterCheckpoint(Map)} instead.
+ *
+ * @param offsets offsets that were checkpointed.
+ */
+ @Deprecated
+ default void onCheckpoint(Map<SystemStreamPartition, String> offsets) {
+ afterCheckpoint(offsets);
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/18ebfce3/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala b/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
index d47458d..f5cc6fd 100644
--- a/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
+++ b/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
@@ -268,18 +268,19 @@ class OffsetManager(
if (checkpointManager != null || checkpointListeners.nonEmpty) {
debug("Getting checkpoint offsets for taskName %s." format taskName)
- val sspsForTaskName = systemStreamPartitions.getOrElse(taskName, throw new SamzaException("No SSPs registered for task: " + taskName)).toSet
- val sspToOffsets = lastProcessedOffsets.get(taskName)
+ val taskStartingOffsets = startingOffsets.getOrElse(taskName,
+ throw new SamzaException("Couldn't find starting offsets for task: " + taskName))
- val partitionOffsets = if (sspToOffsets != null) {
- // Filter the offsets in case the task model changed since the last checkpoint was written.
- sspToOffsets.asScala.filterKeys(sspsForTaskName.contains)
- } else {
- warn(taskName + " is not found... ")
- Map[SystemStreamPartition, String]()
- }
+ val taskSSPs = systemStreamPartitions.getOrElse(taskName,
+ throw new SamzaException("No SSPs registered for task: " + taskName))
- new Checkpoint(new HashMap(partitionOffsets.asJava)) // Copy into new Map to prevent mutation
+ // filter the offsets in case the task model changed since the last checkpoint was written.
+ val taskLastProcessedOffsets =
+ lastProcessedOffsets.getOrDefault(taskName, new ConcurrentHashMap()).asScala
+ .filterKeys(taskSSPs.contains)
+
+ val modifiedTaskOffsets = getModifiedOffsets(taskStartingOffsets, taskLastProcessedOffsets)
+ new Checkpoint(new HashMap(modifiedTaskOffsets)) // Copy into new Map to prevent mutation
} else {
debug("Returning null checkpoint for taskName %s because no checkpoint manager/callback is defined." format taskName)
null
@@ -287,6 +288,47 @@ class OffsetManager(
}
/**
+ * Call the beforeCheckpoint() method on [[CheckpointListener]] SystemConsumers to give them a chance to inspect and
+ * potentially modify the offsets about to be checkpointed.
+ */
+ def getModifiedOffsets(taskStartingOffsets: Map[SystemStreamPartition, String],
+ taskLastProcessedOffsets: Map[SystemStreamPartition, String]): HashMap[SystemStreamPartition, String] = {
+ val modifiedOffsets = new HashMap[SystemStreamPartition, String](taskLastProcessedOffsets.asJava)
+
+ taskLastProcessedOffsets
+ .groupBy { case (ssp, offset) => ssp.getSystem } // Map(systemName -> Map(ssp -> offset))
+ .foreach { case (systemName, systemSSPLastProcessedOffsets) =>
+ if (checkpointListeners.contains(systemName)) {
+ val checkpointListener = checkpointListeners(systemName)
+
+ // if we haven't processed any messages from the system since we started, we don't need to call
+ // the checkpoint listener. also, in case of Kafka, getting the safe offset is only possible after
+ // first poll() is successful. check if we need to adjust the offset for any of the ssp in the system.
+ val needModifiedOffsets = systemSSPLastProcessedOffsets.exists { case (ssp, sspLastProcessedOffset) =>
+ // starting offsets are checkpointed offsets + 1 (see 'loadStartingOffsets'),
+ // while on container start last processed offsets are equal to checkpointed offsets.
+ // compare last processed offsets with starting offsets to see if we've processed anything.
+ // last processed offsets < starting offsets after container start but before we've processed anything.
+ taskStartingOffsets.get(ssp) match {
+ case Some(startingOffset) if startingOffset != null =>
+ val systemAdmin = systemAdmins.getSystemAdmin(systemName)
+ val comparisionResult = systemAdmin.offsetComparator(sspLastProcessedOffset, startingOffset)
+ if (comparisionResult != null) comparisionResult < 0 else false
+ case _ => false
+ }
+ }
+
+ if (needModifiedOffsets) {
+ val systemModifiedOffsets = checkpointListener.beforeCheckpoint(systemSSPLastProcessedOffsets.asJava)
+ modifiedOffsets.putAll(systemModifiedOffsets)
+ }
+ }
+ }
+
+ modifiedOffsets
+ }
+
+ /**
* Write the specified checkpoint for the given task and delete the corresponding startpoint, if available.
*/
def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
http://git-wip-us.apache.org/repos/asf/samza/blob/18ebfce3/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
index 31ff9cf..5e45547 100644
--- a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
+++ b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
@@ -20,6 +20,7 @@
package org.apache.samza.checkpoint
import java.util
+import java.util.function.BiConsumer
import org.apache.samza.{Partition, SamzaException}
import org.apache.samza.config.MapConfig
@@ -31,7 +32,11 @@ import org.apache.samza.system._
import org.apache.samza.util.NoOpMetricsRegistry
import org.junit.Assert._
import org.junit.Test
+import org.mockito.Matchers._
import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
+import org.mockito.{Matchers, Mockito}
import org.scalatest.Assertions.intercept
import scala.collection.JavaConverters._
@@ -410,6 +415,73 @@ class TestOffsetManager {
assertEquals("47", offsetManager.offsetManagerMetrics.checkpointedOffsets.get(systemStreamPartition).getValue)
}
+ @Test
+ def testGetModifiedOffsets: Unit = {
+ val system1 = "system1"
+ val system2 = "system2"
+ val ssp1 = new SystemStreamPartition(system1, "stream", new Partition(1))
+ val ssp2 = new SystemStreamPartition(system1, "stream1", new Partition(1))
+ val ssp3 = new SystemStreamPartition(system1, "stream1", new Partition(2))
+ val ssp4 = new SystemStreamPartition(system1, "stream1", new Partition(3))
+ val ssp5 = new SystemStreamPartition(system2, "stream", new Partition(1))
+ val ssp6 = new SystemStreamPartition(system2, "stream1", new Partition(1))
+
+ val lastProcessedOffsets = Map(ssp1 -> "10", ssp2 -> "20", ssp3 -> "30", ssp4 -> "40", ssp5 -> "50", ssp6 -> "60")
+ // starting offsets are (lastProcessedOffset + 1) (i.e. checkpointed offset + 1) on startup
+ val taskStartingOffsets = Map(ssp1 -> "11", ssp2 -> "19", ssp3 -> null, ssp5 -> "51", ssp6 -> "59")
+
+ // test behavior without any checkpoint listeners
+ val offsetManager = new OffsetManager()
+ val regularOffsets = offsetManager.getModifiedOffsets(taskStartingOffsets, lastProcessedOffsets)
+ // since there are no checkpoint listeners, there should be no change in offsets.
+ assertEquals("10", regularOffsets.get(ssp1))
+ assertEquals("20", regularOffsets.get(ssp2))
+ assertEquals("30", regularOffsets.get(ssp3))
+ assertEquals("40", regularOffsets.get(ssp4))
+ assertEquals("50", regularOffsets.get(ssp5))
+ assertEquals("60", regularOffsets.get(ssp6))
+
+ // test behavior with a checkpoint listener for "system1" that increments all provided offsets by 5
+ val checkpointListener: CheckpointListener = new CheckpointListener {
+ override def beforeCheckpoint(offsets: util.Map[SystemStreamPartition, String]) = {
+ val results = new util.HashMap[SystemStreamPartition, String]()
+ offsets.forEach(new BiConsumer[SystemStreamPartition, String] {
+ override def accept(ssp: SystemStreamPartition, offset: String): Unit = {
+ results.put(ssp, (offset.toLong + 5).toString)
+ }
+ })
+ results
+ }
+ }
+ val checkpointListeners = Map(system1 -> checkpointListener)
+
+ val system1Admin = mock(classOf[SystemAdmin])
+ Mockito.when(system1Admin.offsetComparator(anyString(), anyString()))
+ .thenAnswer(new Answer[Integer] {
+ override def answer(invocation: InvocationOnMock): Integer = {
+ val offset1 = invocation.getArguments.apply(0).asInstanceOf[String]
+ val offset2 = invocation.getArguments.apply(1).asInstanceOf[String]
+ offset1.toLong.compareTo(offset2.toLong)
+ }
+ })
+
+ val systemAdmins = mock(classOf[SystemAdmins])
+ when(systemAdmins.getSystemAdmin(Matchers.eq("system1"))).thenReturn(system1Admin)
+
+ val offsetManagerWithCheckpointListener =
+ new OffsetManager(checkpointListeners = checkpointListeners, systemAdmins = systemAdmins)
+ val modifiedOffsets = offsetManagerWithCheckpointListener.getModifiedOffsets(taskStartingOffsets, lastProcessedOffsets)
+ // since there is at least one ssp on system1 that has processed messages (ssp2),
+ // all ssps on system1 should get modified offsets (ssp1 to ssp4)
+ assertEquals("15", modifiedOffsets.get(ssp1))
+ assertEquals("25", modifiedOffsets.get(ssp2))
+ assertEquals("35", modifiedOffsets.get(ssp3))
+ assertEquals("45", modifiedOffsets.get(ssp4))
+ // no change for ssps on system2
+ assertEquals("50", modifiedOffsets.get(ssp5))
+ assertEquals("60", modifiedOffsets.get(ssp6))
+ }
+
// Utility method to create and write checkpoint in one statement
def checkpoint(offsetManager: OffsetManager, taskName: TaskName): Unit = {
offsetManager.writeCheckpoint(taskName, offsetManager.buildCheckpoint(taskName))