fix ClockService

huafengw · huafengw · commit c08bfb9bc4fe · 2014-10-27T16:38:55.000+08:00
diff --git a/examples/src/main/scala/org/apache/gearpump/streaming/examples/wordcount/Split.scala b/examples/src/main/scala/org/apache/gearpump/streaming/examples/wordcount/Split.scala
@@ -54,6 +54,6 @@ class Split(conf : Configs) extends TaskActor(conf) {
         output(new Message(msg, System.currentTimeMillis()))
       }
     }
-    self ! Message("continue")
+    self ! Message("continue", System.currentTimeMillis())
   }
 }
diff --git a/streaming/src/main/scala/org/apache/gearpump/streaming/task/ClockService.scala b/streaming/src/main/scala/org/apache/gearpump/streaming/task/ClockService.scala
@@ -46,8 +46,8 @@ import org.apache.gearpump.streaming.task.ClockService._
   override def preStart : Unit = {
     dag.tasks.foreach { taskIdWithDescription =>
       val (taskGroupId, description) = taskIdWithDescription
-      val taskClocks = new Array[TimeStamp](description.parallism)
-      val taskgroupClock = new TaskGroupClock(taskGroupId, 0L, taskClocks)
+      val taskClocks = new Array[TimeStamp](description.parallism).map(_ => Long.MaxValue)
+      val taskgroupClock = new TaskGroupClock(taskGroupId, Long.MaxValue, taskClocks)
       taskgroupClocks.add(taskgroupClock)
       taskgroupLookup.put(taskGroupId, taskgroupClock)
     }
@@ -88,7 +88,7 @@ import org.apache.gearpump.streaming.task.ClockService._
 
 object ClockService {
 
-  class TaskGroupClock(val taskgroup : TaskGroup, var minClock : TimeStamp = 0L, var taskClocks : Array[TimeStamp] = null) extends Comparable[TaskGroupClock] {
+  class TaskGroupClock(val taskgroup : TaskGroup, var minClock : TimeStamp = Long.MaxValue, var taskClocks : Array[TimeStamp] = null) extends Comparable[TaskGroupClock] {
     override def equals(obj: Any): Boolean = {
       if (this.eq(obj.asInstanceOf[AnyRef])) {
         return true

Original file line number	Diff line number	Diff line change
`@@ -54,6 +54,6 @@ class Split(conf : Configs) extends TaskActor(conf) {`
`54`	`54`	`output(new Message(msg, System.currentTimeMillis()))`
`55`	`55`	`}`
`56`	`56`	`}`
`57`		`- self ! Message("continue")`
	`57`	`+ self ! Message("continue", System.currentTimeMillis())`
`58`	`58`	`}`
`59`	`59`	`}`