fix ClockService bug

huafengw · huafengw · commit f49e275ec24f · 2014-10-27T13:37:16.000+08:00
diff --git a/core/src/main/scala/org/apache/gearpump/cluster/AppManager.scala b/core/src/main/scala/org/apache/gearpump/cluster/AppManager.scala
@@ -114,7 +114,7 @@ private[cluster] class AppManager() extends Actor with Stash {
   def waitForMasterState: Receive = {
     case GetSuccess(_, replicatedState: GSet, _) =>
       state = replicatedState.getValue().asScala.foldLeft(state) { (set, appState) =>
-         set + appState.asInstanceOf[ApplicationState]
+        set + appState.asInstanceOf[ApplicationState]
       }
       appId = state.map(_.appId).size
       LOG.info(s"Successfully recoeved application states for ${state.map(_.appId)}, nextAppId: ${appId}....")
@@ -158,7 +158,7 @@ private[cluster] class AppManager() extends Actor with Stash {
         case Some(info) =>
           val worker = info.worker
           LOG.info(s"Shuttdown app master at ${worker.path}, appId: $appId, executorId: $masterExecutorId")
-          //cleanApplicationData(appId)
+          cleanApplicationData(appId)
           worker ! ShutdownExecutor(appId, masterExecutorId, s"AppMaster $appId shutdown requested by master...")
           sender ! ShutdownApplicationResult(Success(appId))
         case None =>
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
@@ -47,7 +47,7 @@ import org.apache.gearpump.streaming.task.ClockService._
     dag.tasks.foreach { taskIdWithDescription =>
       val (taskGroupId, description) = taskIdWithDescription
       val taskClocks = new Array[TimeStamp](description.parallism)
-      val taskgroupClock = new TaskGroupClock(taskGroupId, 0, taskClocks)
+      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
