JobInProgress job
core/src/main/java/org/apache/hama/bsp/BSPMaster.java
++totalSubmissions;
    for (JobInProgressListener listener : jobInProgressListeners) {
      try {
        listener.recoverTaskInJob(job);
void recoverTask(JobInProgress job)
action instanceof KillTaskAction
core/src/main/java/org/apache/hama/bsp/GroomServer.java
tasks.remove(killAction.getTaskID());
                runningTasks.remove(killAction.getTaskID());
else if (action instanceof KillTaskAction)
action instanceof RecoverTaskAction
core/src/main/java/org/apache/hama/bsp/GroomServer.java
LOG.info("Recovery action task.");
            RecoverTaskAction recoverAction = (RecoverTaskAction) action;
            Task t = recoverAction.getTask();
            LOG.info("Recovery action task." + t.getTaskID());
            synchronized (assignedPeerNames) {
              prevPort = BSPNetUtils.getNextAvailable(prevPort);
              assignedPeerNames.put(t.getTaskID(), prevPort);
else if (action instanceof RecoverTaskAction)
long superstepNumber
core/src/main/java/org/apache/hama/bsp/GroomServer.java
if (this.taskStatus.getRunState() != TaskStatus.State.FAILED) {
        this.taskStatus.setRunState(TaskStatus.State.RECOVERING);
        this.taskStatus.setPhase(TaskStatus.Phase.RECOVERING);
        this.taskStatus.setStateString("recovering");
void markAsRecoveryTask(long superstepNumber)
TaskAttemptID taskid
core/src/main/java/org/apache/hama/bsp/GroomServer.java
TaskInProgress tip = tasks.get(taskid);
    if (tip != null) {
      return tip.getStatus();
TaskStatus getTaskStatus(TaskAttemptID taskid)
TaskAttemptID taskid
core/src/main/java/org/apache/hama/bsp/GroomServer.java
TaskInProgress tip = tasks.get(taskid);
    if (tip != null) {
      return tip.startSuperstepCount;
long getStartSuperstep(TaskAttemptID taskid)
BSPJobID jobId  Configuration conf
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
this.conf = conf;
    this.jobId = jobId;
    master = null;
public JobInProgress(BSPJobID jobId, Configuration conf)
GroomServerStatus groomStatus  Task task
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
taskToGroomMap.put(task, groomStatus);
    int tasksInGroom = 0;

    if (taskCountInGroomMap.containsKey(groomStatus)) {
      tasksInGroom = taskCountInGroomMap.get(groomStatus);
void updateGroomTaskDetails(GroomServerStatus groomStatus, Task task)
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
return maxTaskAttempts;
int getMaximumReAttempts()
TaskInProgress tip
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
if (this.faultToleranceService == null
        || (!faultToleranceService.isRecoveryPossible(tip)))
      return false;

    if (!faultToleranceService.isAlreadyRecovered(tip)) {
      if(LOG.isDebugEnabled()){
        LOG.debug("Adding recovery task " + tip.getCurrentTaskAttemptId());
boolean handleFailure(TaskInProgress tip)
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
return recoveryTasks.size() != 0;
boolean isRecoveryPending()
core/src/main/java/org/apache/hama/bsp/JobInProgress.java
return this.faultToleranceService;
FaultTolerantMasterService getFaultToleranceService()
BSPMessageBundle<? extends Writable> bundle
core/src/main/java/org/apache/hama/bsp/LocalBSPRunner.java
for (Writable value : bundle.getMessages()) {
        loopBackMessage(value);
void loopBackMessages(BSPMessageBundle<? extends Writable> bundle)
Writable message
core/src/main/java/org/apache/hama/bsp/LocalBSPRunner.java
localIncomingMessages.add((M)message);
      peer.incrementCounter(BSPPeerImpl.PeerCounter.TOTAL_MESSAGES_RECEIVED,
          1L);
void loopBackMessage(Writable message)
String key  Writable writableVal
core/src/main/java/org/apache/hama/bsp/LocalBSPRunner.java
return false;
    @Override
    public boolean remove(String key, SyncEventListener listener) {
      return false;
boolean getInformation(String key, Writable writableVal)
core/src/main/java/org/apache/hama/bsp/SimpleTaskScheduler.java
Map<GroomServerStatus, List<GroomServerAction>> actionMap = new HashMap<GroomServerStatus, List<GroomServerAction>>(
        LOG.error("Could not schedule all tasks!");
        List<GroomServerAction> taskActions = actionMap.get(groomStatus);
          taskActions = new ArrayList<GroomServerAction>(
      sendDirectivesToGrooms(actionMap);

      return Boolean.TRUE;
Boolean scheduleNewTasks()
core/src/main/java/org/apache/hama/bsp/SimpleTaskScheduler.java
// Action to be sent for each task to the respective groom server.
      Map<GroomServerStatus, List<GroomServerAction>> actionMap = new HashMap<GroomServerStatus, List<GroomServerAction>>(
          2 * this.groomStatuses.size());

      try {
        jip.recoverTasks(groomStatuses, actionMap);
Boolean scheduleRecoveryTasks()

        Map<GroomServerStatus  List<GroomServerAction>> actionMap
core/src/main/java/org/apache/hama/bsp/SimpleTaskScheduler.java
while ((jip.getStatus().getRunState() == JobStatus.RUNNING || jip
          .getStatus().getRunState() == JobStatus.RECOVERING)
        List<GroomServerAction> actionList = actionMap.get(groomStatus);
          return Boolean.FALSE;
          && (jip.getStatus().getRunState() != JobStatus.RUNNING || jip
              .getStatus().getRunState() != JobStatus.RECOVERING)) {
        return Boolean.FALSE;

    public Boolean call() {
      if (jip.isRecoveryPending()) {
        return scheduleRecoveryTasks();
Boolean sendDirectivesToGrooms(
        Map<GroomServerStatus, List<GroomServerAction>> actionMap)
TaskAttemptID taskid 
      Map<String  GroomServerStatus> grooms 
      Map<GroomServerStatus  Integer> tasksInGroomMap 
      String[] possibleLocations
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
for (int i = 0; i < possibleLocations.length; ++i) {
      String location = possibleLocations[i];
      GroomServerStatus groom = grooms.get(location);
      if (groom == null)
        continue;
      Integer taskInGroom = tasksInGroomMap.get(groom);
      taskInGroom = (taskInGroom == null) ? 0 : taskInGroom;
      if (taskInGroom < groom.getMaxTasks()
          && location.equals(groom.getGroomHostName())) {
        return groom.getGroomHostName();
String getGroomToSchedule(TaskAttemptID taskid,
      Map<String, GroomServerStatus> grooms,
      Map<GroomServerStatus, Integer> tasksInGroomMap,
      String[] possibleLocations)
Map<String  GroomServerStatus> grooms 
      Map<GroomServerStatus  Integer> tasksInGroomMap
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
Iterator<String> groomIter = grooms.keySet().iterator();
    while (groomIter.hasNext()) {
      GroomServerStatus groom = grooms.get(groomIter.next());
      if (groom == null)
        continue;
      Integer taskInGroom = tasksInGroomMap.get(groom);
      taskInGroom = (taskInGroom == null) ? 0 : taskInGroom;
      if (taskInGroom < groom.getMaxTasks()) {
        return groom.getGroomHostName();
String getAnyGroomToSchedule(Map<String, GroomServerStatus> grooms,
      Map<GroomServerStatus, Integer> tasksInGroomMap)
GroomServerStatus groomStatus
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
if(groomStatus == null){
      return null;
Task constructTask(GroomServerStatus groomStatus)
TaskAttemptID taskid 
  // Map<String  GroomServerStatus> grooms 
  // Map<GroomServerStatus  Integer> tasksInGroomMap
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
// String splitClass = null;
  // BytesWritable split = null;
  // Task t = null;
  // if (rawSplit != null) {
  // splitClass = rawSplit.getClassName();
  // split = rawSplit.getBytes();
  // String[] possibleLocations = rawSplit.getLocations();
  // String groomName = getGroomToSchedule(taskid, grooms, tasksInGroomMap,
  // possibleLocations);
  // if (groomName != null) {
  // t = new BSPTask(jobId, jobFile, taskid, partition, splitClass, split);
  // activeTasks.put(taskid, groomName);
  // myGroomStatus = grooms.get(groomName);
  //
Task getGroomForTask(TaskAttemptID taskid,
  // Map<String, GroomServerStatus> grooms,
  // Map<GroomServerStatus, Integer> tasksInGroomMap)
TaskAttemptID taskid 
      Map<String  GroomServerStatus> grooms 
      Map<GroomServerStatus  Integer> tasksInGroomMap 
      String[] possibleLocations
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
String splitClass = null;
    BytesWritable split = null;
    String groomName = getGroomToSchedule(taskid, grooms, tasksInGroomMap,
        possibleLocations);
    if (groomName != null) {
      t = new BSPTask(jobId, jobFile, taskid, partition, splitClass, split);
      activeTasks.put(taskid, groomName);
      myGroomStatus = grooms.get(groomName);
Task getGroomForRecoverTaskInHosts(TaskAttemptID taskid,
      Map<String, GroomServerStatus> grooms,
      Map<GroomServerStatus, Integer> tasksInGroomMap,
      String[] possibleLocations)
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return (nextTaskId < (MAX_TASK_EXECS + maxTaskAttempts));
boolean canStartTask()
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return taskid;
  // /** Remove */
  // public Task getTaskToRun(Map<String, GroomServerStatus> grooms,
  // Map<GroomServerStatus, Integer> tasksInGroomMap) throws IOException {
  // TaskAttemptID taskId = computeTaskId();
  // if (taskId == null) {
  // return null;
  //
TaskAttemptID computeTaskId()
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return mySuperstep;
int getSuperstep()
int mySuperstep
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
this.mySuperstep = mySuperstep;
void setSuperstep(int mySuperstep)
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return this.rawSplit;
RawSplit getFileSplit()
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return this.currentTaskId;
TaskAttemptID getCurrentTaskAttemptId()
String peerName  M message
core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
Iterator<MessageEventListener<M>> iterator = this.messageListenerQueue
        .iterator();
    while (iterator.hasNext()) {
      iterator.next().onMessageSent(peerName, message);
void notifySentMessage(String peerName, M message)
core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
Iterator<MessageEventListener<M>> iterator = this.messageListenerQueue
        .iterator();
    while (iterator.hasNext()) {
      iterator.next().onInitialized();
void notifyInit()
core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
Iterator<MessageEventListener<M>> iterator = this.messageListenerQueue
        .iterator();
    while (iterator.hasNext()) {
      iterator.next().onClose();
void notifyClose()
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncBSPMasterClient.java
return this.zk;
ZooKeeper getZK()
String key
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
InterruptedException {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Checking node " + path);
String correctKey(String key)
String path 
      Writable valueHolder
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
try {
          getValueFromBytes(data, valueHolder);
boolean extractData(String path,
      Writable valueHolder)
String key  Writable valueHolder
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
return extractData(path, valueHolder);
  /**
   * Clears all sub-children of node bspRoot
   */
  protected void clearZKNodes() {
    try {
      Stat s = zk.exists(bspRoot, false);
      if (s != null) {
        clearZKNodes(bspRoot);
boolean getInformation(String key, Writable valueHolder)
WatchedEvent arg0
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
void process(WatchedEvent arg0)
String key  SyncEventListener listener
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
key = correctKey(key);
    try {
      clearZKNodes(key);
      this.zk.delete(key, -1);
      return true;
boolean remove(String key, SyncEventListener listener)
TaskAttemptID attemptId  BSPPeer<?  ?  ?  ?  Text> peer 
        Configuration conf  InetSocketAddress peerAddress
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
// TODO Auto-generated method stub
void init(TaskAttemptID attemptId, BSPPeer<?, ?, ?, ?, Text> peer,
        Configuration conf, InetSocketAddress peerAddress)
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
// TODO Auto-generated method stub
void close()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void clearOutgoingQueues()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return this.messageQueue.size();
int getNumCurrentMessages()
BSPMessageBundle<? extends Writable> bundle
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
this.loopbackBundle = (BSPMessageBundle<Text>) bundle;
void loopBackMessages(BSPMessageBundle<? extends Writable> bundle)
Writable message
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void loopBackMessage(Writable message)
BSPJob job  Configuration conf  TaskAttemptID taskId 
        Counters counters  long superstep  BSPPeerSyncClient syncClient 
        MessageManager<Text> messenger  TaskStatus.State state
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
this.conf = conf;
      if (superstep > 0)
        superstepCount = superstep;
      else
        superstepCount = 0L;

      try {
        fService = (new AsyncRcvdMsgCheckpointImpl<Text>()).constructPeerFaultTolerance(
            job, (BSPPeer<?, ?, ?, ?, Text>) this,
            (BSPPeerSyncClient) syncClient, null, taskId, superstep, conf,
            messenger);
        this.fService.onPeerInitialized(state);
public TestBSPPeer(BSPJob job, Configuration conf, TaskAttemptID taskId,
        Counters counters, long superstep, BSPPeerSyncClient syncClient,
        MessageManager<Text> messenger, TaskStatus.State state)
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return 1;
int getNumCurrentMessages()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return superstepCount;
long getSuperstepCount()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return null;
String getPeerName()
int index
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return null;
String getPeerName(int index)
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return 1;
int getPeerIndex()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return 0;
int getNumPeers()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void clear()
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return null;
Configuration getConfiguration()
Enum<?> name
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return null;
Counter getCounter(Enum<?> name)
String group  String name
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return null;
Counter getCounter(String group, String name)
Enum<?> key  long amount
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void incrementCounter(Enum<?> key, long amount)
String group  String counter  long amount
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void incrementCounter(String group, String counter, long amount)
BSPJobID jobId  String... args
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
StringBuffer buffer = new StringBuffer(100);
      buffer.append(jobId.toString()).append("/");
      for (String arg : args) {
        buffer.append(arg).append("/");
String constructKey(BSPJobID jobId, String... args)
String key  Writable value 
        boolean permanent  SyncEventListener listener
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
ArrayWritable writables = (ArrayWritable) value;
      long step = ((LongWritable) writables.get()[0]).get();
      long count = ((LongWritable) writables.get()[1]).get();

      LOG.info("SyncClient Storing value step = " + step + " count = " + count
          + " for key " + key);
      valueMap.put(key, value);
      return true;
boolean storeInformation(String key, Writable value,
        boolean permanent, SyncEventListener listener)
String key 
        Writable valueHolder
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
LOG.info("Getting value for key " + key);
      if(!valueMap.containsKey(key)){
        return false;
boolean getInformation(String key,
        Writable valueHolder)
String key  boolean permanent 
        SyncEventListener listener
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
valueMap.put(key, NullWritable.get());
      return true;
boolean addKey(String key, boolean permanent,
        SyncEventListener listener)
String key
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return valueMap.containsKey(key);
boolean hasKey(String key)
String key  SyncEvent event 
        SyncEventListener listener
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
return false;
boolean registerListener(String key, SyncEvent event,
        SyncEventListener listener)
String key  SyncEventListener listener
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
valueMap.remove(key);
      return false;
boolean remove(String key, SyncEventListener listener)
BSPJobID jobId  TaskAttemptID taskId 
        String hostAddress  long port
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void register(BSPJobID jobId, TaskAttemptID taskId,
        String hostAddress, long port)
BSPJobID jobId  TaskAttemptID taskId 
        String hostAddress  long port
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void deregisterFromBarrier(BSPJobID jobId, TaskAttemptID taskId,
        String hostAddress, long port)
core/src/test/java/org/apache/hama/bsp/TestCheckpoint.java
void stopServer()
core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java
checkPointInterval = conf.getInt(Constants.CHECKPOINT_INTERVAL, 1);
    if (LOG.isDebugEnabled())
      LOG.debug(new StringBuffer(1000).append("Enabled = ")
          .append(conf.getBoolean(Constants.CHECKPOINT_ENABLED, false))
          .append(" checkPointInterval = ").append(checkPointInterval)
          .append(" lastCheckPointStep = ").append(lastCheckPointStep)
          .append(" getSuperstepCount() = ").append(getSuperstepCount())
          .toString());
    return (conf.getBoolean(Constants.CHECKPOINT_ENABLED, false)
        && (checkPointInterval != 0) && (((int) (getSuperstepCount() - lastCheckPointStep)) >= checkPointInterval));
  private final String checkpointedPath() {
    String backup = conf.get("bsp.checkpoint.prefix_path", "/checkpoint/");
    String ckptPath = backup + bspJob.getJobID().toString() + "/"
        + getSuperstepCount() + "/" + this.taskId.toString();
    if (LOG.isDebugEnabled())
      LOG.debug("Messages are to be saved to " + ckptPath);
    return ckptPath;
  final void checkpoint(String checkpointedPath, BSPMessageBundle<M> bundle) {
    FSDataOutputStream out = null;
    try {
      out = this.fs.create(new Path(checkpointedPath));
      bundle.write(out);
boolean isReadyToCheckpoint()
BSPMaster bspMaster
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
this.bspMaster = bspMaster;
void setBspMaster(BSPMaster bspMaster)
core/src/main/java/org/apache/hama/bsp/TaskInProgress.java
return bspMaster;
BSPMaster getBspMaster()
BSPMessageBundle<M> messages
core/src/main/java/org/apache/hama/bsp/message/AvroMessageManagerImpl.java
peer.incrementCounter(BSPPeerImpl.PeerCounter.TOTAL_MESSAGES_RECEIVED,
        messages.getMessages().size());
    Iterator<M> iterator = messages.getMessages().iterator();
    while (iterator.hasNext()) {
      this.localQueueForNextIteration.add(iterator.next());
      iterator.remove();
void put(BSPMessageBundle<M> messages)
M msg
core/src/main/java/org/apache/hama/bsp/message/HadoopMessageManagerImpl.java
this.localQueueForNextIteration.add(msg);
    peer.incrementCounter(BSPPeerImpl.PeerCounter.TOTAL_MESSAGES_RECEIVED, 1L);
  public final void put(BSPMessageBundle<M> messages) {
    for (M message : messages.getMessages()) {
      this.localQueueForNextIteration.add(message);
void put(M msg)
BSPCompressedBundle compMsgBundle
core/src/main/java/org/apache/hama/bsp/message/HadoopMessageManagerImpl.java
for (M message : bundle.getMessages()) {
      this.localQueueForNextIteration.add(message);
void put(BSPCompressedBundle compMsgBundle)
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncBSPMasterClient.java
try {
      Stat s = zk.exists(bspRoot, false);
      if (s != null) {
        clearZKNodes(bspRoot);
void clearZKNodes()
String key
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
InterruptedException {
   * @return The instance of Writable object. 
  protected Writable getValueFromBytes(
      byte[] data, Class<? extends Writable> classType) throws IOException{
    Writable value = null;
      value = ReflectionUtils
          .newInstance(classType, new Object[0]);
        value.readFields(diStream);
String correctKey(String key)
String path 
      Class<? extends Writable> classType
core/src/main/java/org/apache/hama/bsp/sync/ZKSyncClient.java
Writable value = null;
        try{
          value = getValueFromBytes(data, classType);
Writable extractData(String path,
      Class<? extends Writable> classType)
core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java
public String getValue(){
    
      // TODO Auto-generated method stub
      
      
      // TODO Auto-generated method stub
      
    
      // TODO Auto-generated method stub
    final PeerSyncClient syncClient = (PeerSyncClient)
    		SyncServiceFactory.getPeerSyncClient(conf);
    try {
      IntWritable data = new IntWritable(5);
      syncClient.storeInformation(
          syncClient.constructKey(jobId, String.valueOf(1L), "test"), data,
          true, null);
      
      ListenerTest listenerTest = new ListenerTest();
      
      
      syncClient.registerListener(
          syncClient.constructKey(jobId, String.valueOf(1L), "test"), 
          ZKSyncEventFactory.getValueChangeEvent(),
          listenerTest);
      
      IntWritable value = (IntWritable) syncClient.getInformation(
          syncClient.constructKey(jobId, String.valueOf(1L), "test"),
          IntWritable.class);
      assertTrue(value != null);
      int intVal = value == null ? 0 : value.get();
      assertTrue(intVal == data.get());
      
      data.set(6);
      syncClient.storeInformation(
          syncClient.constructKey(jobId, String.valueOf(1L), "test"), data,
          true, null);
      value = (IntWritable) syncClient.getInformation(
          syncClient.constructKey(jobId, String.valueOf(1L), "test"),
          IntWritable.class);

      
      intVal = value == null ? 0 : value.get();
      assertTrue(intVal == data.get());
      
      Thread.sleep(5000);
      
      assertEquals(true, listenerTest.getValue().equals("Changed"));
      
      
      syncServer.stopServer();
public ListenerTest()