diff --git a/core/src/main/java/org/apache/hama/bsp/taskallocation/RoundRobinTaskAllocator.java b/core/src/main/java/org/apache/hama/bsp/taskallocation/RoundRobinTaskAllocator.java
new file mode 100644
index 0000000..0bf060b
--- /dev/null
+++ b/core/src/main/java/org/apache/hama/bsp/taskallocation/RoundRobinTaskAllocator.java
@@ -0,0 +1,146 @@
+/**
+ * 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.hama.bsp.taskallocation;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hama.bsp.GroomServerStatus;
+import org.apache.hama.bsp.TaskInProgress;
+import org.apache.hama.bsp.BSPJobClient.RawSplit;
+
+/**
+ * <code>RoundRobinTaskAllocator</code> is a round robin based task allocator that equally
+ * divides the tasks among all the Grooms. It balances the load of cluster. For example
+ * if a cluster has 10 Grooms and 20 tasks are to be scheduled then each Groom which
+ * get 2 tasks.
+ */
+public class RoundRobinTaskAllocator implements TaskAllocationStrategy {
+
+  Log LOG = LogFactory.getLog(RoundRobinTaskAllocator.class);
+
+  @Override
+  public void initialize(Configuration conf) {
+  }
+
+  /**
+   * This function loops through the whole list of Grooms with their task count
+   * and returns the first Groom which contains the minimum number of tasks.
+   * @param groomStatuses The map of groom-name to
+   *          <code>GroomServerStatus</code> object for all known grooms.
+   * @param taskCountInGroomMap Map of count of tasks in groom (To be deprecated
+   *          soon)
+   * @return returns the groom name which should be allocated the next task or 
+   *          null no suitable groom was found.
+   */
+  private String findGroomWithMinimumTasks(
+      Map<String, GroomServerStatus> groomStatuses,
+      Map<GroomServerStatus, Integer> taskCountInGroomMap) {
+    
+    Entry<GroomServerStatus, Integer> firstGroomWithMinimumTasks = null;
+    
+    // At the start taskCountInGroomMap is empty so we have to put 0 tasks on grooms
+    if (taskCountInGroomMap.size() < groomStatuses.size()) {
+      for (String s : groomStatuses.keySet()) {
+        GroomServerStatus groom = groomStatuses.get(s);
+        if (groom == null)
+          continue;
+        Integer taskInGroom = taskCountInGroomMap.get(groom);
+        
+        // Find the groom that is yet to get its first tasks and assign 0 value to it.
+        // Having zero will make sure that it gets selected.
+        if (taskInGroom == null) {
+          taskCountInGroomMap.put(groom, 0);
+          break;
+        }
+      }
+    }
+    
+    for (Entry<GroomServerStatus, Integer> currentGroom : taskCountInGroomMap.entrySet()) {
+      if (firstGroomWithMinimumTasks == null || firstGroomWithMinimumTasks.getValue() > currentGroom.getValue()) {
+        if(currentGroom.getValue() < currentGroom.getKey().getMaxTasks()) { // Assign the task to groom which still has space for more tasks 
+          firstGroomWithMinimumTasks = currentGroom;
+        } // If there is no space then continue and find the next best groom
+      }
+    }
+    
+    return (firstGroomWithMinimumTasks == null) ? null
+        : firstGroomWithMinimumTasks.getKey().getGroomHostName();
+  }
+  
+  /**
+   * Select grooms that has the block of data locally stored on the groom
+   * server.
+   */
+  @Override
+  public String[] selectGrooms(Map<String, GroomServerStatus> groomStatuses,
+      Map<GroomServerStatus, Integer> taskCountInGroomMap,
+      BSPResource[] resources, TaskInProgress taskInProgress) {
+    if (!taskInProgress.canStartTask()) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Cannot start task based on id");
+      }
+      return new String[0];
+    }
+
+    RawSplit rawSplit = taskInProgress.getFileSplit();
+    if (rawSplit != null) {
+      return rawSplit.getLocations();
+    }
+    return null;
+  }
+
+  @Override
+  public GroomServerStatus getGroomToAllocate(
+      Map<String, GroomServerStatus> groomStatuses, String[] selectedGrooms,
+      Map<GroomServerStatus, Integer> taskCountInGroomMap,
+      BSPResource[] resources, TaskInProgress taskInProgress) {
+    if (!taskInProgress.canStartTask()) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Exceeded allowed attempts.");
+      }
+      return null;
+    }    
+
+    String groomName = null;
+
+    groomName = findGroomWithMinimumTasks(groomStatuses, taskCountInGroomMap);
+
+    if (groomName != null) {
+      return groomStatuses.get(groomName);
+    }
+    
+    return null;
+  }
+
+  /**
+   * This operation is not supported.
+   */
+  @Override
+  public Set<GroomServerStatus> getGroomsToAllocate(
+      Map<String, GroomServerStatus> groomStatuses, String[] selectedGrooms,
+      Map<GroomServerStatus, Integer> taskCountInGroomMap,
+      BSPResource[] resources, TaskInProgress taskInProgress) {
+    throw new UnsupportedOperationException(
+        "This API is not supported for the called API function call.");
+  }
+}
diff --git a/core/src/test/java/org/apache/hama/bsp/TestTaskAllocationRoundRobin.java b/core/src/test/java/org/apache/hama/bsp/TestTaskAllocationRoundRobin.java
new file mode 100644
index 0000000..e471fe0
--- /dev/null
+++ b/core/src/test/java/org/apache/hama/bsp/TestTaskAllocationRoundRobin.java
@@ -0,0 +1,128 @@
+/**
+ * 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.hama.bsp;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.junit.Test;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hama.bsp.BSPJobClient.RawSplit;
+import org.apache.hama.bsp.taskallocation.BSPResource;
+import org.apache.hama.bsp.taskallocation.RoundRobinTaskAllocator;
+import org.apache.hama.bsp.taskallocation.TaskAllocationStrategy;
+
+public class TestTaskAllocationRoundRobin extends TestCase {
+
+  public static final Log LOG = LogFactory
+      .getLog(TestTaskAllocationRoundRobin.class);
+
+  Configuration conf = new Configuration();
+  Map<String, GroomServerStatus> groomStatuses;
+  Map<GroomServerStatus, Integer> taskCountInGroomMap;
+  BSPResource[] resources;
+  TaskInProgress taskInProgress;
+
+  @Override
+  protected void setUp() throws Exception {
+    super.setUp();
+
+    String[] locations = new String[] { "host6", "host4", "host3" };
+    String value = "data";
+    RawSplit split = new RawSplit();
+    split.setLocations(locations);
+    split.setBytes(value.getBytes(), 0, value.getBytes().length);
+    split.setDataLength(value.getBytes().length);
+
+    assertEquals(value.getBytes().length, (int) split.getDataLength());
+
+    taskCountInGroomMap = new LinkedHashMap<GroomServerStatus, Integer>(10);
+    resources = new BSPResource[0];
+    BSPJob job = new BSPJob(new BSPJobID("checkpttest", 1), "/tmp");
+    JobInProgress jobProgress = new JobInProgress(job.getJobID(), conf);
+    taskInProgress = new TaskInProgress(job.getJobID(), "job.xml", split, conf,
+        jobProgress, 1);
+
+    groomStatuses = new LinkedHashMap<String, GroomServerStatus>(10);
+
+    for (int i = 0; i < 10; ++i) {
+      String name = "host" + i;
+
+      GroomServerStatus status = new GroomServerStatus(name,
+          new ArrayList<TaskStatus>(), 0, 3, "", name);
+      groomStatuses.put(name, status);
+      taskCountInGroomMap.put(status, 0);
+    }
+  }
+
+  @Override
+  protected void tearDown() throws Exception {
+    super.tearDown();
+  }
+
+  /**
+   * This test simulates the allocation of 30 tasks in round robin fashion
+   * on 10 Grooms.
+   */
+  @Test
+  public void testRoundRobinAllocation() {
+    TaskAllocationStrategy strategy = ReflectionUtils.newInstance(conf
+        .getClass("", RoundRobinTaskAllocator.class,
+            TaskAllocationStrategy.class), conf);
+
+    for (int i = 0; i < 30; i++) {
+      GroomServerStatus groomStatus = strategy.getGroomToAllocate(
+          groomStatuses, null, taskCountInGroomMap, resources, taskInProgress);
+      if (groomStatus != null) {
+        taskCountInGroomMap.put(groomStatus,
+            taskCountInGroomMap.get(groomStatus) + 1); // Increment the total tasks in it
+        
+        assertEquals("", "host" + (i % 10), groomStatus.getGroomHostName()); 
+      }
+    }
+  }
+
+  @Test
+  public void testRoundRobinDataLocality() throws Exception {
+
+    TaskAllocationStrategy strategy = ReflectionUtils.newInstance(conf
+        .getClass("", RoundRobinTaskAllocator.class,
+            TaskAllocationStrategy.class), conf);
+
+    String[] hosts = strategy.selectGrooms(groomStatuses, taskCountInGroomMap,
+        resources, taskInProgress);
+
+    List<String> list = new ArrayList<String>();
+
+    for (int i = 0; i < hosts.length; ++i) {
+      list.add(hosts[i]);
+    }
+
+    assertTrue(list.contains("host6"));
+    assertTrue(list.contains("host3"));
+    assertTrue(list.contains("host4"));
+  }
+
+}