diff --git a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointManager.java b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointManager.java
index 4e76bb2..4dcefaa 100644
--- a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointManager.java
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointManager.java
@@ -19,8 +19,6 @@
 
 package org.apache.samza.checkpoint;
 
-import java.util.HashMap;
-import java.util.Map;
 import org.apache.samza.container.TaskName;
 
 /**
@@ -68,15 +66,4 @@
    * Clear the checkpoints in the checkpoint stream.
    */
   default void clearCheckpoints() { }
-
-  /**
-   * Returns the last recorded checkpoint for all tasks present in the implementation-specific location.
-   * All tasks contains all the tasks within the current job model.
-   * All tasks also includes tasks which may have been part of the job model during a previous deploy.
-   * @return A Map of TaskName to Checkpoint object.
-   *         The Checkpoint object has the recorded offset data of the specified partition.
-   */
-  default Map<TaskName, Checkpoint> readAllCheckpoints() {
-    return new HashMap<>();
-  };
 }
diff --git a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
index 1f4a740..3908170 100644
--- a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
+++ b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
@@ -132,9 +132,7 @@
     if (envelopeKeyorOffset == null) {
       return new SystemStreamPartition(systemStreamPartition, 0);
     }
-    // modulo 31 first to best spread out the hashcode and then modulo elasticityFactor for actual keyBucket
-    // Note: elasticityFactor <= 16 so modulo 31 is safe to do.
-    int keyBucket = (Math.abs(envelopeKeyorOffset.hashCode()) % 31) % elasticityFactor;
+    int keyBucket = Math.abs(envelopeKeyorOffset.hashCode()) % elasticityFactor;
     return new SystemStreamPartition(systemStreamPartition, keyBucket);
   }
 
diff --git a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
index d36ac66..b9aa82c 100644
--- a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
@@ -190,6 +190,7 @@
   public static final String CONTAINER_HEARTBEAT_MONITOR_ENABLED = "job.container.heartbeat.monitor.enabled";
   private static final boolean CONTAINER_HEARTBEAT_MONITOR_ENABLED_DEFAULT = true;
 
+
   // Enabled elasticity for the job
   // number of (elastic) tasks in the job will be old task count X elasticity factor
   public static final String JOB_ELASTICITY_FACTOR = "job.elasticity.factor";
@@ -517,8 +518,8 @@
 
   public int getElasticityFactor() {
     int elasticityFactor = getInt(JOB_ELASTICITY_FACTOR, DEFAULT_JOB_ELASTICITY_FACTOR);
-    if (elasticityFactor < 1 || elasticityFactor > 16) {
-      throw new ConfigException("Elasticity factor can not be less than 1 or greater than 16");
+    if (elasticityFactor < 1) {
+      throw new ConfigException("Elasticity factor can not be less than 1");
     }
     return elasticityFactor;
   }
diff --git a/samza-core/src/main/java/org/apache/samza/container/RunLoop.java b/samza-core/src/main/java/org/apache/samza/container/RunLoop.java
index 5bbcbae..3334adc 100644
--- a/samza-core/src/main/java/org/apache/samza/container/RunLoop.java
+++ b/samza-core/src/main/java/org/apache/samza/container/RunLoop.java
@@ -860,19 +860,8 @@
         IncomingMessageEnvelope envelope = pendingEnvelope.envelope;
 
         if (envelope.isEndOfStream()) {
-          if (elasticityFactor <= 1) {
-            SystemStreamPartition ssp = envelope.getSystemStreamPartition();
-            processingSspSet.remove(ssp);
-          } else {
-            // if envelope is end of stream, the ssp of envelope should be removed from task's processing set irresp of keyBucket
-            SystemStreamPartition sspOfEnvelope = envelope.getSystemStreamPartition();
-            Optional<SystemStreamPartition> ssp = processingSspSet.stream()
-                .filter(sspInSet -> sspInSet.getSystemStream().equals(sspOfEnvelope.getSystemStream())
-                    && sspInSet.getPartition().equals(sspOfEnvelope.getPartition()))
-                .findFirst();
-            ssp.ifPresent(processingSspSet::remove);
-            ssp.ifPresent(processingSspSetToDrain::remove);
-          }
+          SystemStreamPartition ssp = envelope.getSystemStreamPartition(elasticityFactor);
+          processingSspSet.remove(ssp);
           if (!hasIntermediateStreams) {
             pendingEnvelopeQueue.remove();
           }
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupByPartition.java b/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupByPartition.java
index 53e7400..1efd172 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupByPartition.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupByPartition.java
@@ -64,7 +64,7 @@
         int keyBucket = elasticityFactor == 1 ? -1 : i;
         String taskNameStr = elasticityFactor == 1 ?
             String.format("Partition %d", ssp.getPartition().getPartitionId()) :
-            String.format("Partition %d_%d_%d", ssp.getPartition().getPartitionId(), keyBucket, elasticityFactor);
+            String.format("Partition %d %d", ssp.getPartition().getPartitionId(), keyBucket);
         TaskName taskName = new TaskName(taskNameStr);
         SystemStreamPartition sspWithKeyBucket = new SystemStreamPartition(ssp, keyBucket);
         groupedMap.putIfAbsent(taskName, new HashSet<>());
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupBySystemStreamPartition.java b/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupBySystemStreamPartition.java
index 652cf15..9443048 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupBySystemStreamPartition.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupBySystemStreamPartition.java
@@ -63,8 +63,7 @@
         SystemStreamPartition sspWithKeyBucket = new SystemStreamPartition(ssp, keyBucket);
         HashSet<SystemStreamPartition> sspSet = new HashSet<SystemStreamPartition>();
         sspSet.add(sspWithKeyBucket);
-        String elasticitySuffix = elasticityFactor == 1 ? "" : String.format("_%d", elasticityFactor);
-        groupedMap.put(new TaskName(sspWithKeyBucket.toString() + elasticitySuffix), sspSet);
+        groupedMap.put(new TaskName(sspWithKeyBucket.toString()), sspSet);
       }
     }
 
diff --git a/samza-core/src/main/java/org/apache/samza/elasticity/util/ElasticityUtils.java b/samza-core/src/main/java/org/apache/samza/elasticity/util/ElasticityUtils.java
deleted file mode 100644
index 90c3748..0000000
--- a/samza-core/src/main/java/org/apache/samza/elasticity/util/ElasticityUtils.java
+++ /dev/null
@@ -1,493 +0,0 @@
-/*
- * 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.samza.elasticity.util;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.samza.checkpoint.Checkpoint;
-import org.apache.samza.container.TaskName;
-import org.apache.samza.system.SystemAdmin;
-import org.apache.samza.system.SystemAdmins;
-import org.apache.samza.system.SystemStreamPartition;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Class with util methods to be used for checkpoint computation when elasticity is enabled
- * Elasticity is supported only  for tasks created by either
- * the {@link org.apache.samza.container.grouper.stream.GroupByPartition} SSP grouper or
- * the {@link org.apache.samza.container.grouper.stream.GroupBySystemStreamPartition} SSP grouper
- */
-public class ElasticityUtils {
-  private static final Logger log = LoggerFactory.getLogger(ElasticityUtils.class);
-
-  // GroupByPartition tasks have names like Partition 0_1_2
-  // where 0 is the partition number, 1 is the key bucket and 2 is the elasticity factor
-  // see {@link GroupByPartition.ELASTIC_TASK_NAME_FORMAT}
-  private static final String ELASTIC_TASK_NAME_GROUP_BY_PARTITION_REGEX = "Partition (\\d+)_(\\d+)_(\\d+)";
-  private static final Pattern ELASTIC_TASK_NAME_GROUP_BY_PARTITION_PATTERN = Pattern.compile(ELASTIC_TASK_NAME_GROUP_BY_PARTITION_REGEX);
-  private static final String TASK_NAME_GROUP_BY_PARTITION_REGEX = "Partition (\\d+)";
-  private static final Pattern TASK_NAME_GROUP_BY_PARTITION_PATTERN = Pattern.compile(TASK_NAME_GROUP_BY_PARTITION_REGEX);
-  private static final String TASK_NAME_GROUP_BY_PARTITION_PREFIX = "Partition ";
-
-  //GroupBySSP tasks have names like "SystemStreamPartition [<system>, <Stream>, <partition>, keyBucket]_2"
-  // where 2 is the elasticity factor
-  // see {@link GroupBySystemStreamPartition} and {@link SystemStreamPartition.toString}
-  private static final String ELASTIC_TASK_NAME_GROUP_BY_SSP_REGEX = "SystemStreamPartition \\[(\\S+), (\\S+), (\\d+), (\\d+)\\]_(\\d+)";
-  private static final Pattern ELASTIC_TASK_NAME_GROUP_BY_SSP_PATTERN = Pattern.compile(ELASTIC_TASK_NAME_GROUP_BY_SSP_REGEX);
-  private static final String TASK_NAME_GROUP_BY_SSP_REGEX = "SystemStreamPartition \\[(\\S+), (\\S+), (\\d+)\\]";
-  private static final Pattern TASK_NAME_GROUP_BY_SSP_PATTERN = Pattern.compile(TASK_NAME_GROUP_BY_SSP_REGEX);
-  private static final String TASK_NAME_GROUP_BY_SSP_PREFIX = "SystemStreamPartition ";
-
-  /**
-   * Elasticity is supported for GroupByPartition tasks and GroupBySystemStreamPartition tasks
-   * When elasticity is enabled, GroupByPartition tasks have names Partition 0_1_2
-   * When elasticity is enabled, GroupBySystemStreamPartition tasks have names SystemStreamPartition [systemA, streamB, 0, 1]_2
-   * Both tasks have names ending with _%d where %d is the elasticity factor
-   * @param taskName of either GroupByPartition or GroupBySystemStreamPartition task
-   * @return
-   *   for GroupByPartition and GroupBySystemStreamPartition tasks returns elasticity factor from the task name
-   *   for other tasks returns 1 which is the default elasticity factor
-   */
-  static int getElasticityFactorFromTaskName(TaskName taskName) {
-    return getTaskNameParts(taskName).elasticityFactor;
-  }
-
-  /**
-   * checks if the given taskname is of a GroupByPartition task
-   * @param taskName of any task
-   * @return true if GroupByPartition (starts with prefix "Partition ") or false otherwise
-   */
-  static boolean isGroupByPartitionTask(TaskName taskName) {
-    return taskName.getTaskName().startsWith(TASK_NAME_GROUP_BY_PARTITION_PREFIX);
-  }
-
-  /**
-   * checks if the given taskname is of a GroupBySystemStreamPartition task
-   * @param taskName of any task
-   * @return true if GroupBySystemStreamPartition (starts with prefix "SystemStreamPartition ") or false otherwise
-   */
-  static boolean isGroupBySystemStreamPartitionTask(TaskName taskName) {
-    return taskName.getTaskName().startsWith(TASK_NAME_GROUP_BY_SSP_PREFIX);
-  }
-
-  /**
-   * checks if given taskName is elastic aka created with an elasticity factor > 1
-   * @param taskName of any task
-   * @return true for following, false otherwise
-   *    for task created by GroupByPartition, taskName has format "Partition 0_1_2"
-   *    for task created by GroupBySystemStreamPartition, taskName has format "SystemStreamPartition [systemA, streamB, 0, 1]_2"
-   */
-  static boolean isTaskNameElastic(TaskName taskName) {
-    if (isGroupByPartitionTask(taskName)) {
-      Matcher m = ELASTIC_TASK_NAME_GROUP_BY_PARTITION_PATTERN.matcher(taskName.getTaskName());
-      return m.find();
-    } else if (isGroupBySystemStreamPartitionTask(taskName)) {
-      Matcher m = ELASTIC_TASK_NAME_GROUP_BY_SSP_PATTERN.matcher(taskName.getTaskName());
-      return m.find();
-    }
-    return false;
-  }
-
-  /**
-   * From given taskName extract the values for system, stream, partition, keyBucket and elasticityFactor
-   * @param taskName any taskName
-   * @return TaskNameComponents object containing system, stream, partition, keyBucket and elasticityFactor
-   *    for GroupByPartition task:
-   *         taskNames are of the format "Partition 0_1_2" (with elasticity) or "Partition 0" (without elasticity)
-   *         system and stream are empty "" strings and partition is the input partition,
-   *         without elasticity, keyBucket = 0 and elasticityFactor = 1 (the default values)
-   *         with elasticity, keyBucket from name (ex 1 above) and elasticityFactor (ex 2 from above)
-   *    for GroupBySystemStreamPartition task:
-   *         taskNames are of the format "SystemStreamPartition [systemA, streamB, 0, 1]_2" (with elasticity) or
-   *         "SystemStreamPartition [systemA, streamB, 0]" (without elasticity)
-   *         system and stream and partition are from the name (ex system = systemA, steram = streamB, partition =0 above)
-   *         without elasticity, keyBucket = 0 and elasticityFactor = 1 (the default values)
-   *         with elasticity, keyBucket from name (ex 1 above) and elasticityFactor (ex 2 from above)
-   *   for tasks created with other SSP groupers:
-   *        default TaskNameComponents is returned which has empty system, stream,
-   *        -1 for partition and 0 for keyBucket and 1 for elasticity factor
-   */
-  static TaskNameComponents getTaskNameParts(TaskName taskName) {
-    if (isGroupByPartitionTask(taskName)) {
-      return getTaskNameParts_GroupByPartition(taskName);
-    } else if (isGroupBySystemStreamPartitionTask(taskName)) {
-      return getTaskNameParts_GroupBySSP(taskName);
-    }
-    log.warn("TaskName {} is neither GroupByPartition nor GroupBySystemStreamPartition task. "
-        + "Elasticity is not supported for this taskName. "
-        + "Returning default TaskNameComponents which has default keyBucket 0,"
-        + " default elasticityFactor 1 and invalid partition -1", taskName.getTaskName());
-    return new TaskNameComponents(TaskNameComponents.INVALID_PARTITION);
-  }
-
-  /**
-   * see doc for getTaskNameParts above
-   */
-  private static TaskNameComponents getTaskNameParts_GroupByPartition(TaskName taskName) {
-    String taskNameStr = taskName.getTaskName();
-    log.debug("GetTaskNameParts for taskName {}", taskNameStr);
-
-    Matcher matcher = ELASTIC_TASK_NAME_GROUP_BY_PARTITION_PATTERN.matcher(taskNameStr);
-    if (matcher.find()) {
-      return new TaskNameComponents(Integer.valueOf(matcher.group(1)),
-          Integer.valueOf(matcher.group(2)),
-          Integer.valueOf(matcher.group(3)));
-    }
-    matcher = TASK_NAME_GROUP_BY_PARTITION_PATTERN.matcher(taskNameStr);
-    if (matcher.find()) {
-      return new TaskNameComponents(Integer.valueOf(matcher.group(1)));
-    }
-    log.error("Could not extract partition, keybucket and elasticity factor from taskname for task {}.", taskNameStr);
-    throw new IllegalArgumentException("TaskName format incompatible");
-  }
-
-  /**
-   * see doc for getTaskNameParts above
-   */
-  private static TaskNameComponents getTaskNameParts_GroupBySSP(TaskName taskName) {
-    String taskNameStr = taskName.getTaskName();
-    log.debug("GetTaskNameParts for taskName {}", taskNameStr);
-
-    Matcher matcher = ELASTIC_TASK_NAME_GROUP_BY_SSP_PATTERN.matcher(taskNameStr);
-    if (matcher.find()) {
-      return new TaskNameComponents(matcher.group(1),
-          matcher.group(2),
-          Integer.valueOf(matcher.group(3)),
-          Integer.valueOf(matcher.group(4)),
-          Integer.valueOf(matcher.group(5)));
-    }
-    matcher = TASK_NAME_GROUP_BY_SSP_PATTERN.matcher(taskNameStr);
-    if (matcher.find()) {
-      return new TaskNameComponents(matcher.group(1),
-          matcher.group(2),
-          Integer.valueOf(matcher.group(3)));
-    }
-    log.warn("Could not extract system, stream, partition, keybucket and elasticity factor from taskname for task {}.", taskNameStr);
-    throw new IllegalArgumentException("TaskName format incompatible");
-  }
-
-  /**
-   * Without elasticity, a task consumes an entire (full) SSP = [System, stream, partition].
-   * With elasticity, a task consumes a portion of the SSP_withKeyBucket = [system, stream, partition, keyBucket]
-   *    where 0 <= keyBucket < elasticityFactor and contains a subset of the IncomingMessageEnvelope(IME) from the full SSP
-   * Given two tasks currentTask and otherTask, the task otherTask is called ancestor of currentTask if the following is true
-   *    all IME consumed by currentTask will be consumed by otherTask when elasticityFactor decreases or stays same
-   *    For example:
-   *      case 1: elasticityFactor 2 to 1
-   *            otherTask = Partition 0 consuming all IME in SSP = [systemA, streamB, 0] when elasticityFactor=1
-   *            currentTask1 = Partition 0_0_2 consumes IME in SSP_withKeyBucket0 = [systemA, streamB, 0, 0 (keyBucket)] when elasticityFactor = 2
-   *            currentTask2 = Partition 0_1_2 consumes IME in SSP_withKeyBucket1 = [systemA, streamB, 0, 1 (keyBucket)] when elasticityFactor = 2
-   *            SSP =  SSP_withKeyBucket0 + SSP_withKeyBucket1. Thus, Partition 0 is ancestor of Partition 0_0_2 and Partition 0_1_2
-   *      case 2: elasticityFactor 2 to 2 - no change
-   *            Partition 0_0_2 is an ancestor of itself since the input SSP_withKeyBucket0 doesnt change
-   *            similarly Partition 0_1_2 is an ancestor of itself. This applies to all elasticityFactors
-   *      case 3: elasticityFactor 4 to 2
-   *            otherTask = Partition 0_0_2 consuming all IME in SSP_withKeyBucket0 = [systemA, streamB, 0, 0] when elasticityFactor=2
-   *            currentTask1 = Partition 0_0_4 consumes IME in SSP_withKeyBucket00 = [systemA, streamB, 0, 0 (keyBucket)] when elasticityFactor = 4
-   *            currentTask2 = Partition 0_2_4 consumes IME in SSP_withKeyBucket01 = [systemA, streamB, 0, 2 (keyBucket)] when elasticityFactor = 4
-   *            From the computation of SSP_withkeyBucket in {@link org.apache.samza.system.IncomingMessageEnvelope}
-   *            we have getSystemStreamPartition(int elasticityFactor) which does keyBucket = (Math.abs(envelopeKeyorOffset.hashCode()) % 31) % elasticityFactor;
-   *            Thus, SSP_withKeyBucket0 = SSP_withKeyBucket00 + SSP_withKeyBucket01.
-   *            Thus, Partition 0_0_2 is ancestor of Partition 0_0_4 and Partition 0_2_4
-   *            Similarly, Partition 0_1_2 is ancestor of Partition 0_1_4 and Partition 0_3_4
-   *            And transitively, Partition 0 is ancestor of Partition 0_0_4, Partition 0_1_4, Partition 0_2_4 and Partition 0_3_4
-   *
-   * This applies to tasks created by GroupByPartition and GroupBySystemStreamPartition SSPGroupers.
-   * aka this applies if both currentTask and otherTask are created by GroupByPartition or both are created by GroupBySystemStreamPartition
-   * If either currentTask and/or otherTask were created by other SSPGroupers then false is returned.
-   * @param currentTask
-   * @param otherTask
-   * @return true if otherTask is ancestor of currentTask, false otherwise
-   */
-  static boolean isOtherTaskAncestorOfCurrentTask(TaskName currentTask, TaskName otherTask) {
-    log.debug("isOtherTaskAncestorOfCurrentTask with currentTask {} and otherTask {}", currentTask, otherTask);
-    if (!((isGroupByPartitionTask(currentTask) && isGroupByPartitionTask(otherTask))
-        || (isGroupBySystemStreamPartitionTask(currentTask) && isGroupBySystemStreamPartitionTask(otherTask)))) {
-      return false;
-    }
-
-    TaskNameComponents currentTaskNameComponents = getTaskNameParts(currentTask);
-    TaskNameComponents otherTaskNameComponents = getTaskNameParts(otherTask);
-
-    if (!otherTaskNameComponents.system.equals(currentTaskNameComponents.system)
-        || !otherTaskNameComponents.stream.equals(currentTaskNameComponents.stream)
-        || otherTaskNameComponents.partition != currentTaskNameComponents.partition
-        || otherTaskNameComponents.elasticityFactor > currentTaskNameComponents.elasticityFactor) {
-      return false;
-    }
-
-    return (currentTaskNameComponents.keyBucket % otherTaskNameComponents.elasticityFactor) == otherTaskNameComponents.keyBucket;
-  }
-
-  /**
-   * See javadoc for isOtherTaskAncestorOfCurrentTask above
-   * Given currentTask and otherTask,
-   *   if currentTask == otherTask, then its not a descendant. (unlike ancestor)
-   *   else, if isOtherTaskAncestorOfCurrentTask(otherTask, currentTask) then otherTask is descendant of currentTask
-   * @param currentTask
-   * @param otherTask
-   * @return
-   */
-  static boolean isOtherTaskDescendantOfCurrentTask(TaskName currentTask, TaskName otherTask) {
-    log.debug("isOtherTaskDescendantOfCurrentTask with currentTask {} and otherTask {}", currentTask, otherTask);
-    if (!((isGroupByPartitionTask(currentTask) && isGroupByPartitionTask(otherTask))
-        || (isGroupBySystemStreamPartitionTask(currentTask) && isGroupBySystemStreamPartitionTask(otherTask)))) {
-      return false;
-    }
-
-    TaskNameComponents currentTaskNameComponents = getTaskNameParts(currentTask);
-    TaskNameComponents otherTaskNameComponents = getTaskNameParts(otherTask);
-
-    if (!otherTaskNameComponents.system.equals(currentTaskNameComponents.system)
-        || !otherTaskNameComponents.stream.equals(currentTaskNameComponents.stream)
-        || otherTaskNameComponents.partition != currentTaskNameComponents.partition
-        || otherTaskNameComponents.elasticityFactor <= currentTaskNameComponents.elasticityFactor) {
-      return false;
-    }
-
-    return (
-        otherTaskNameComponents.keyBucket % currentTaskNameComponents.elasticityFactor) == currentTaskNameComponents.keyBucket;
-  }
-
-  /**
-   * For a given taskName and a map of task names to checkpoints, returns the taskName's ancestor and descendants checkpoints
-   * All ancestor checkpoints are put into a set
-   * Descendant checkpoins are put into a map of elasticityFactor to descendant checkpoint where the elastictyFactor is of the descendant.
-   * For example, given taskName Partition 0_0_2 and checkpoint Map (Partition 0->C1, Partition 0_0_4-> C2, Partition 0_1_4 -> C3, Partition 0_2_4 ->C4)
-   * the return value is AncestorSet = <C1> and descendantMap = (4 -> <C2, C4>)
-   * See javadoc of isOtherTaskAncestorOfCurrentTask and isOtherTaskDescendantOfCurrentTask for definition of ancestor and descendant
-   * @param taskName name of the task
-   * @param checkpointMap map from taskName to checkpoint
-   * @return Pair of AncestorCheckpoint set and Descendant Checkpoint Map
-   */
-  static Pair<Set<Checkpoint>, Map<Integer, Set<Checkpoint>>> getAncestorAndDescendantCheckpoints(
-      TaskName taskName, Map<TaskName, Checkpoint> checkpointMap) {
-    Set<Checkpoint> ancestorCheckpoints = new HashSet<>();
-    Map<Integer, Set<Checkpoint>> descendantCheckpoints = new HashMap<>();
-    log.debug("starting to parse the checkpoint map to find ancestors and descendants for taskName {}", taskName.getTaskName());
-    checkpointMap.keySet().forEach(otherTaskName -> {
-      Checkpoint otherTaskCheckpoint = checkpointMap.get(otherTaskName);
-      if (isOtherTaskAncestorOfCurrentTask(taskName, otherTaskName)) {
-        log.debug("current task name is {} and other task name is {} and other task is ancestor", taskName, otherTaskName);
-        ancestorCheckpoints.add(otherTaskCheckpoint);
-      }
-      if (isOtherTaskDescendantOfCurrentTask(taskName, otherTaskName)) {
-        log.debug("current task name is {} and other task name is {} and other task is descendant", taskName, otherTaskName);
-        int otherEF = getElasticityFactorFromTaskName(otherTaskName);
-        if (!descendantCheckpoints.containsKey(otherEF)) {
-          descendantCheckpoints.put(otherEF, new HashSet<>());
-        }
-        descendantCheckpoints.get(otherEF).add(otherTaskCheckpoint);
-      }
-    });
-    log.debug("done computing all ancestors and descendants of {}", taskName);
-    return new ImmutablePair<>(ancestorCheckpoints, descendantCheckpoints);
-  }
-
-  /**
-   * Given a checkpoint with offset map from SystemStreamPartition to offset, returns the offset for the desired ssp
-   * Only the system, stream and partition portions of the SSP are matched, the keyBucket is not considered.
-   * A checkpoint belongs to one task and a task would consume either the full SSP (aka no keyBucket)
-   * or consume exactly one of the keyBuckets of an SSP. Hence there will be at most one entry for an SSP in a checkpoint
-   * @param checkpoint Checkpoint containing SSP -> offset
-   * @param ssp SystemStreamPartition for which an offset needs to be fetched
-   * @return offset for the ssp in the Checkpoint or null if doesnt exist.
-   */
-  static String getOffsetForSSPInCheckpoint(Checkpoint checkpoint, SystemStreamPartition ssp) {
-    String checkpointStr = checkpoint.getOffsets().entrySet().stream()
-        .map(k -> k.getKey() + " : " + k.getValue())
-        .collect(Collectors.joining(", ", "{", "}"));
-    log.debug("for ssp {}, in checkpoint {}", ssp, checkpointStr);
-
-    Optional<String> offsetFound = checkpoint.getOffsets().entrySet()
-        .stream()
-        .filter(entry -> entry.getKey().getSystemStream().equals(ssp.getSystemStream()) && entry.getKey()
-            .getPartition()
-            .equals(ssp.getPartition()))
-        .map(Map.Entry::getValue)
-        .findFirst();
-    if (offsetFound.isPresent()) {
-      return offsetFound.get();
-    }
-    log.warn("Could not find offset for ssp {} in checkpoint {}. returning null string as offset", ssp, checkpoint);
-    return null;
-  }
-
-  /**
-   * Given a set of checkpoints, find the max aka largest offset for an ssp
-   * Largest is determined by the SystemAdmin.offsetCompartor of the ssp's system.
-   * Only the system, stream and partition portions of the SSP are matched, the keyBucket is not considered.
-   * @param checkpointSet set of checkpoints
-   * @param ssp for which largest offset is needed
-   * @param systemAdmin of the ssp.getSystem()
-   * @return offset - string if one exists else null
-   */
-  static String getMaxOffsetForSSPInCheckpointSet(Set<Checkpoint> checkpointSet,
-      SystemStreamPartition ssp, SystemAdmin systemAdmin) {
-    return checkpointSet.stream()
-        .filter(Objects::nonNull)
-        .map(checkpoint -> getOffsetForSSPInCheckpoint(checkpoint, ssp))
-        .filter(Objects::nonNull)
-        .sorted((offset1, offset2) -> systemAdmin.offsetComparator(offset2, offset1)) //confirm reverse sort - aka largest offset first
-        .findFirst().orElse(null);
-  }
-
-  /**
-   * Given a set of checkpoints, find the min aka smallest offset for an ssp
-   * Smallest is determined by the SystemAdmin.offsetCompartor of the ssp's system.
-   * Only the system, stream and partition portions of the SSP are matched, the keyBucket is not considered.
-   * @param checkpointSet set of checkpoints
-   * @param ssp for which largest offset is needed
-   * @param systemAdmin of the ssp.getSystem()
-   * @return offset - string if one exists else null
-   */
-  static String getMinOffsetForSSPInCheckpointSet(Set<Checkpoint> checkpointSet,
-      SystemStreamPartition ssp, SystemAdmin systemAdmin) {
-    return checkpointSet.stream()
-        .filter(Objects::nonNull)
-        .map(checkpoint -> getOffsetForSSPInCheckpoint(checkpoint, ssp))
-        .filter(Objects::nonNull)
-        .sorted((offset1, offset2) -> systemAdmin.offsetComparator(offset1, offset2)) //confirm ascending sort - aka smallest offset first
-        .findFirst().orElse(null);
-  }
-
-  /**
-   * Prereq: See javadoc for isOtherTaskAncestorOfCurrentTask and isOtherTaskDescendantOfCurrentTask to fully understand ancestor and descendant notion
-   * Briefly, Given tasks - Partition 0, Partition 0_0_2, Partition 0_1_2 and Partition 0_0_4, Partition 0_1_4, Partition 0_2_4 and Partition 0_3_4
-   * (recall Partition 0_1_2 means reads input partition 0, keyBucket 1 and elasticityFactor 2)
-   * For task Partition 0_0_2: ancestors = [Partition 0, Partition 0_0_2] and descendants = [Partition 0_0_4, Partition 0_2_4]
-   *
-   * If a task has no descendants, then we just need to pick the largest offset among all the ancestors to get the last processed offset.
-   * for example above, if Partition 0_0_2 only had ancestors and no descendants, taking largest offset among Partition 0 and 0_0_2 gives last proc offset.
-   *
-   * With descendants, a little care is needed. there could be descendants with different elasticity factors.
-   * given one elasticity factor, each the descendant within the elasticity factor consumes a sub-portion (aka keyBucket) of the task.
-   * hence, to avoid data loss, we need to pick the lowest offset across descendants of the same elasticity factor.
-   * Across elasticity factors, largest works just like in ancestor
-   *
-   * Taking a concrete example
-   * From {@link org.apache.samza.system.IncomingMessageEnvelope} (IME)
-   *    Partition 0 consunmig all IME in SSP = [systemA, streamB, 0] when elasticityFactor=1
-   *    Partition 0_1_2 consuming all IME in SSP_withKeyBucket0 = [systemA, streamB, 0, 1 (keyBucket)] when elasticityFactor=2
-   *    Partition 0_0_2 consuming all IME in SSP_withKeyBucket1 = [systemA, streamB, 0, 0 (keyBucket)] when elasticityFactor=2
-   *    Partition 0_0_4 consumes IME in SSP_withKeyBucket00 = [systemA, streamB, 0, 0 (keyBucket)] when elasticityFactor = 4
-   *    Partition 0_2_4 consumes IME in SSP_withKeyBucket01 = [systemA, streamB, 0, 2 (keyBucket)] when elasticityFactor = 4
-   *    From the computation of SSP_withkeyBucket in {@link org.apache.samza.system.IncomingMessageEnvelope}
-   *    we have getSystemStreamPartition(int elasticityFactor) which does keyBucket = (Math.abs(envelopeKeyorOffset.hashCode()) % 31) % elasticityFactor;
-   *    Thus,
-   *       SSP = SSP_withKeyBucket0 + SSP_withKeyBucket1.
-   *       SSP_withKeyBucket0 = SSP_withKeyBucket00 + SSP_withKeyBucket01.
-   *    If the checkpoint map has
-   *      Partition 0: (SSP : 1), Partition 0_0_2: (SSP0 : 2), Partition 0_1_2: (SSP1 : 3), Partition 0_0_4: (SSP0 : 4), Partition 0_2_4: (SSP1 : 6)
-   *      looking at these map and knowing that offsets are monotonically increasing, it is clear that last deploy was with elasticity factor = 4
-   *      to get checkpoint for Partition 0_0_2, we need to consider last deploy's offsets.
-   *      picking 6 (offset for Partition 0_2_4) means that 0_0_2 will start proc from 6 but offset 5 was never processed.
-   *      hence we need to take min of offsets within an elasticity factor.
-   *
-   * Given checkpoints for all the tasks in the checkpoint stream,
-   * computing the last proc offset for an ssp checkpoint for a task,
-   * the following needs to be met.
-   *    1. Ancestors: we need to take largest offset among ancestors for an ssp
-   *    2. Descendants:
-   *         a. group descendants by their elasticityFactor.
-   *         b. among descendants of the same elasticityFactor, take the smallest offset for an ssp
-   *         c. once step b is done, we have (elasticityFactor : smallest-offset-for-ssp) set, pick the largest in this set
-   *    3. Pick the larger among the offsets received from step 1 (for ancestors) and step 2 (for descendants)
-   *
-   * @param taskName
-   * @param taskSSPSet
-   * @param checkpointMap
-   * @param systemAdmins
-   * @return
-   */
-  public static Map<SystemStreamPartition, String> computeLastProcessedOffsetsFromCheckpointMap(
-      TaskName taskName,
-      Set<SystemStreamPartition> taskSSPSet,
-      Map<TaskName, Checkpoint> checkpointMap,
-      SystemAdmins systemAdmins) {
-    Pair<Set<Checkpoint>, Map<Integer, Set<Checkpoint>>> acnestorsAndDescendantsFound =
-        getAncestorAndDescendantCheckpoints(taskName, checkpointMap);
-    Set<Checkpoint> ancestorCheckpoints = acnestorsAndDescendantsFound.getLeft();
-    Map<Integer, Set<Checkpoint>> descendantCheckpoints = acnestorsAndDescendantsFound.getRight();
-
-    Map<SystemStreamPartition, String> taskSSPOffsets = new HashMap<>();
-
-    taskSSPSet.forEach(ssp_withKeyBucket -> {
-      log.info("for taskName {} and ssp of the task {}, finding its last proc offset", taskName, ssp_withKeyBucket);
-
-      SystemStreamPartition ssp = new SystemStreamPartition(ssp_withKeyBucket.getSystemStream(),
-          ssp_withKeyBucket.getPartition());
-
-      SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(ssp.getSystem());
-
-      String currentLastOffsetForSSP = null;
-
-      String ancestorLastOffsetForSSP = getMaxOffsetForSSPInCheckpointSet(ancestorCheckpoints, ssp, systemAdmin);
-
-      log.info("for taskName {} and ssp {} got lastoffset from ancestors as {}",
-          taskName, ssp_withKeyBucket, ancestorLastOffsetForSSP);
-
-      String descendantLastOffsetForSSP = descendantCheckpoints.entrySet().stream()
-          .map(entry -> getMinOffsetForSSPInCheckpointSet(entry.getValue(), ssp, systemAdmin)) // at each ef level, find min offset
-          .sorted((offset1, offset2) -> systemAdmin.offsetComparator(offset2, offset1)) //confirm reverse sort - aka largest offset first
-          .filter(Objects::nonNull)
-          .findFirst().orElse(null);
-
-      log.info("for taskName {} and ssp {} got lastoffset from descendants as {}",
-          taskName, ssp_withKeyBucket, descendantLastOffsetForSSP);
-
-      Integer offsetComparison = systemAdmin.offsetComparator(ancestorLastOffsetForSSP, descendantLastOffsetForSSP);
-      if (descendantLastOffsetForSSP == null || (offsetComparison != null && offsetComparison > 0)) { // means ancestorLastOffsetForSSP > descendantLastOffsetForSSP
-        currentLastOffsetForSSP = ancestorLastOffsetForSSP;
-      } else {
-        currentLastOffsetForSSP = descendantLastOffsetForSSP;
-      }
-      if (currentLastOffsetForSSP == null) {
-        log.info("for taskName {} and ssp {} got lastoffset as null. "
-            + "skipping adding this ssp to task's offsets loaded from previous checkpoint", taskName, ssp_withKeyBucket);
-      } else {
-        log.info("for taskName {} and ssp {} got lastoffset as {}", taskName, ssp_withKeyBucket, currentLastOffsetForSSP);
-        taskSSPOffsets.put(ssp_withKeyBucket, currentLastOffsetForSSP);
-      }
-    });
-
-    String checkpointStr = taskSSPOffsets.entrySet().stream()
-        .map(k -> k.getKey() + " : " + k.getValue())
-        .collect(Collectors.joining(", ", "{", "}"));
-    log.info("for taskName {}, returning checkpoint as {}", taskName, checkpointStr);
-    return taskSSPOffsets;
-  }
-
-  public static boolean wasElasticityEnabled(Map<TaskName, Checkpoint> checkpointMap) {
-    return checkpointMap.keySet().stream()
-        .filter(ElasticityUtils::isTaskNameElastic) // true if the taskName has elasticityFactor in it
-        .findFirst().isPresent();
-  }
-}
diff --git a/samza-core/src/main/java/org/apache/samza/elasticity/util/TaskNameComponents.java b/samza-core/src/main/java/org/apache/samza/elasticity/util/TaskNameComponents.java
deleted file mode 100644
index 4b76b84..0000000
--- a/samza-core/src/main/java/org/apache/samza/elasticity/util/TaskNameComponents.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.samza.elasticity.util;
-
-/**
- * POJO class to store system, stream, partition, and keyBucket information associated with a Task,
- *  that is encoded in the task's name.
- */
-public class TaskNameComponents {
-
-  public static final int DEFAULT_KEY_BUCKET = 0;
-  public static final int DEFAULT_ELASTICITY_FACTOR = 1;
-  public static final int INVALID_PARTITION = -1;
-
-  public final String system;
-  public final String stream;
-  public final int partition;
-  public final int keyBucket;
-  public final int elasticityFactor;
-
-  public TaskNameComponents(int partition) {
-    this(partition, DEFAULT_KEY_BUCKET, DEFAULT_ELASTICITY_FACTOR);
-  }
-
-  public TaskNameComponents(int partition, int keyBucket, int elasticityFactor) {
-    this("", "", partition, keyBucket, elasticityFactor);
-  }
-
-  public TaskNameComponents(String system, String stream, int partition) {
-    this(system, stream, partition, DEFAULT_KEY_BUCKET, DEFAULT_ELASTICITY_FACTOR);
-  }
-
-  public TaskNameComponents(String system, String stream, int partition, int keyBucket, int elasticityFactor) {
-    this.system = system;
-    this.stream = stream;
-    this.partition = partition;
-    this.keyBucket = keyBucket;
-    this.elasticityFactor = elasticityFactor;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) return true;
-    if (!(o instanceof TaskNameComponents)) return false;
-
-    TaskNameComponents that = (TaskNameComponents) o;
-
-    if (!(this.system.equals(that.system))
-        || !(this.stream.equals(that.stream))
-        || (this.partition != that.partition)
-        || (this.keyBucket != that.keyBucket)
-        || (this.elasticityFactor != that.elasticityFactor)) {
-      return false;
-    }
-    return true;
-  }
-  @Override
-  public int hashCode() {
-    final int prime = 31;
-    int result = 1;
-    result = prime * result + system.hashCode();
-    result = prime * result + stream.hashCode();
-    result = prime * result + partition;
-    result = prime * result + keyBucket;
-    result = prime * result + elasticityFactor;
-    return result;
-  }
-}
diff --git a/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java b/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
index 3470fd8..e04edee 100644
--- a/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
+++ b/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
@@ -231,9 +231,10 @@
     public Object deserializeKey(String sspString, DeserializationContext ctxt) throws IOException {
       String[] parts = sspString.split("\\.");
       if (parts.length < 3) {
-        throw new IllegalArgumentException("System stream partition expected in format 'system.stream.partition' ");
+        throw new IllegalArgumentException("System stream partition expected in format 'system.stream.partition");
       }
-      return new SystemStreamPartition(new SystemStream(parts[0], parts[1]), new Partition(Integer.parseInt(parts[2])));
+      return new SystemStreamPartition(
+          new SystemStream(parts[0], parts[1]), new Partition(Integer.parseInt(parts[2])));
     }
   }
 
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 311dc6a..48b681e 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
@@ -22,13 +22,12 @@
 import java.util
 import java.util.HashMap
 import java.util.concurrent.ConcurrentHashMap
+
 import org.apache.commons.lang3.StringUtils
 import org.apache.samza.SamzaException
 import org.apache.samza.annotation.InterfaceStability
-import org.apache.samza.checkpoint.OffsetManager.info
-import org.apache.samza.config.{Config, JobConfig, StreamConfig, SystemConfig}
+import org.apache.samza.config.{Config, StreamConfig, SystemConfig}
 import org.apache.samza.container.TaskName
-import org.apache.samza.elasticity.util.ElasticityUtils
 import org.apache.samza.startpoint.{Startpoint, StartpointManager}
 import org.apache.samza.system.SystemStreamMetadata.OffsetType
 import org.apache.samza.system._
@@ -106,9 +105,7 @@
           // Build OffsetSetting so we can create a map for OffsetManager.
           (systemStream, OffsetSetting(systemStreamMetadata, defaultOffsetType, resetOffset))
       }.toMap
-
-    new OffsetManager(offsetSettings, checkpointManager, startpointManager, systemAdmins, checkpointListeners,
-      offsetManagerMetrics)
+    new OffsetManager(offsetSettings, checkpointManager, startpointManager, systemAdmins, checkpointListeners, offsetManagerMetrics)
   }
 }
 
@@ -215,26 +212,16 @@
    * Set the last processed offset for a given SystemStreamPartition.
    */
   def update(taskName: TaskName, systemStreamPartition: SystemStreamPartition, offset: String) {
-    // without elasticity enabled, there is exactly one entry of an ssp in the systemStreamPartitions map for a taskName
-    // with elasticity enabled, there is exactly one of the keyBuckets of an ssp that a task consumes
-    // and hence exactly one entry of an ssp with keyBucket in in the systemStreamPartitions map for a taskName
-    // hence from the given ssp, find its sspWithKeybucket for the task and use that for updating lastProcessedOffsets
-    val sspWithKeyBucket = systemStreamPartitions.getOrElse(taskName,
-      throw new SamzaException("No SSPs registered for task: " + taskName))
-      .filter(ssp => ssp.getSystemStream.equals(systemStreamPartition.getSystemStream)
-        && ssp.getPartition.equals(systemStreamPartition.getPartition))
-      .toIterator.next()
-
     lastProcessedOffsets.putIfAbsent(taskName, new ConcurrentHashMap[SystemStreamPartition, String]())
     taskSSPsWithProcessedOffsetUpdated.putIfAbsent(taskName, new ConcurrentHashMap[SystemStreamPartition, Boolean]())
 
     if (offset != null) {
       if (!offset.equals(IncomingMessageEnvelope.END_OF_STREAM_OFFSET)) {
-        lastProcessedOffsets.get(taskName).put(sspWithKeyBucket, offset)
+        lastProcessedOffsets.get(taskName).put(systemStreamPartition, offset)
       }
       // Record the spp that have received the new messages. The startpoint for each ssp should only be deleted when the
       // ssp has received the new messages. More details in SAMZA-2749.
-      taskSSPsWithProcessedOffsetUpdated.get(taskName).putIfAbsent(sspWithKeyBucket, true)
+      taskSSPsWithProcessedOffsetUpdated.get(taskName).putIfAbsent(systemStreamPartition, true)
     }
   }
 
@@ -500,19 +487,12 @@
 
     val checkpoint = checkpointManager.readLastCheckpoint(taskName)
 
-    val checkpointMap = checkpointManager.readAllCheckpoints()
-    if (!ElasticityUtils.wasElasticityEnabled(checkpointMap)) {
-      if (checkpoint != null) {
-        return Map(taskName -> checkpoint.getOffsets.asScala.toMap)
-      } else {
-        info("Did not receive a checkpoint for taskName %s. Proceeding without a checkpoint." format taskName)
-        return Map(taskName -> Map())
-      }
+    if (checkpoint != null) {
+      Map(taskName -> checkpoint.getOffsets.asScala.toMap)
+    } else {
+      info("Did not receive a checkpoint for taskName %s. Proceeding without a checkpoint." format taskName)
+      Map(taskName -> Map())
     }
-    info("There was elasticity enabled in one of the previous deploys." +
-      "Last processed offsets computation at container start will use elasticity checkpoints if available.")
-    Map(taskName -> ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(taskName,
-      systemStreamPartitions.get(taskName).get.asJava, checkpointMap, systemAdmins).asScala)
   }
 
   /**
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index 606f32a..75f4a8d 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -226,7 +226,7 @@
     // if elasticity is enabled aka elasticity factor > 1
     // then this TaskInstance processes only those envelopes whose key falls
     // within the keyBucket of the SSP assigned to the task.
-    val incomingMessageSsp = getIncomingMessageSSP(envelope)
+    val incomingMessageSsp = envelope.getSystemStreamPartition(elasticityFactor)
 
     if (!ssp2CaughtupMapping.getOrElse(incomingMessageSsp,
       throw new SamzaException(incomingMessageSsp + " is not registered!"))) {
@@ -566,7 +566,7 @@
     // if elasticity is enabled aka elasticity factor > 1
     // then this TaskInstance handles only those envelopes whose key falls
     // within the keyBucket of the SSP assigned to the task.
-    var incomingMessageSsp = getIncomingMessageSSP(envelope)
+    val incomingMessageSsp = envelope.getSystemStreamPartition(elasticityFactor)
 
     if (IncomingMessageEnvelope.END_OF_STREAM_OFFSET.equals(envelope.getOffset)) {
       ssp2CaughtupMapping(incomingMessageSsp) = true
@@ -629,28 +629,4 @@
 
     startingOffset
   }
-
-  private def getIncomingMessageSSP(envelope: IncomingMessageEnvelope): SystemStreamPartition = {
-    if (elasticityFactor <= 1) {
-      return envelope.getSystemStreamPartition
-    }
-    // if elasticityFactor > 1, find the SSP with keyBucket
-    var incomingMessageSsp = envelope.getSystemStreamPartition(elasticityFactor)
-
-    // if envelope is end of stream or watermark or drain,
-    // it needs to be routed to all tasks consuming the ssp irresp of keyBucket
-    val messageType = MessageType.of(envelope.getMessage)
-    if (envelope.isEndOfStream()
-      || envelope.isDrain()
-      || messageType == MessageType.END_OF_STREAM
-      || messageType == MessageType.WATERMARK) {
-
-      incomingMessageSsp = systemStreamPartitions
-        .filter(ssp => ssp.getSystemStream.equals(incomingMessageSsp.getSystemStream)
-          && ssp.getPartition.equals(incomingMessageSsp.getPartition))
-        .toIterator.next()
-      debug("for watermark or end-of-stream or drain envelope, found incoming ssp as {}".format(incomingMessageSsp))
-    }
-    incomingMessageSsp
-  }
 }
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
index e69a973..ba42264 100644
--- a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
@@ -40,7 +40,7 @@
   // Serialize checkpoint as maps keyed by the SSP.toString() to the another map of the constituent SSP components
   // and offset. Jackson can't automatically serialize the SSP since it's not a POJO and this avoids
   // having to wrap it another class while maintaining readability.
-  // { "SSP.toString()" -> {"system": system, "stream": stream, "partition": partition, "keyBucket": keyBucket, "offset": offset)}
+  // { "SSP.toString()" -> {"system": system, "stream": stream, "partition": partition, "offset": offset)}
   def fromBytes(bytes: Array[Byte]): CheckpointV1 = {
     try {
       val jMap = jsonMapper.readValue(bytes, classOf[util.HashMap[String, util.HashMap[String, String]]])
@@ -55,6 +55,7 @@
         require(partition != null, "Partition must be present in JSON-encoded SystemStreamPartition")
         val offset = sspInfo.get("offset")
         // allow null offsets, e.g. for changelog ssps
+
         new SystemStreamPartition(system, stream, new Partition(partition.toInt)) -> offset
       }
 
diff --git a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
index f56af29..a7458db 100644
--- a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
+++ b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
@@ -707,16 +707,6 @@
     }
     assertTrue(exceptionCaught);
 
-    jobConfig =
-        new JobConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_ELASTICITY_FACTOR, Integer.toString(17))));
-    exceptionCaught = false;
-    try {
-      jobConfig.getElasticityFactor();
-    } catch (ConfigException e) {
-      exceptionCaught = true;
-    }
-    assertTrue(exceptionCaught);
-
     jobConfig = new JobConfig(new MapConfig());
     assertEquals(JobConfig.DEFAULT_JOB_ELASTICITY_FACTOR, jobConfig.getElasticityFactor());
   }
diff --git a/samza-core/src/test/java/org/apache/samza/container/TestRunLoop.java b/samza-core/src/test/java/org/apache/samza/container/TestRunLoop.java
index e584974..16ef93d 100644
--- a/samza-core/src/test/java/org/apache/samza/container/TestRunLoop.java
+++ b/samza-core/src/test/java/org/apache/samza/container/TestRunLoop.java
@@ -225,39 +225,6 @@
   }
 
   @Test
-  public void testEndOfStreamElasticityEnabled() {
-
-    TaskName taskName0 = new TaskName(p0.toString() + " 0");
-    TaskName taskName1 = new TaskName(p0.toString() + " 1");
-    SystemStreamPartition ssp = new SystemStreamPartition("testSystem", "testStreamA", p0);
-    SystemStreamPartition ssp0 = new SystemStreamPartition("testSystem", "testStreamA", p0, 0);
-    SystemStreamPartition ssp1 = new SystemStreamPartition("testSystem", "testStreamA", p0, 1);
-
-    // create EOS IME such that its ssp keybucket maps to ssp0 and not to ssp1
-    // task in the runloop should give this ime to both it tasks
-    IncomingMessageEnvelope envelopeEOS = spy(IncomingMessageEnvelope.buildEndOfStreamEnvelope(ssp));
-    when(envelopeEOS.getSystemStreamPartition(2)).thenReturn(ssp0);
-
-
-    // two task in the run loop that processes ssp0 -> 0th keybucket of ssp and ssp1 -> 1st keybucket of ssp
-    // EOS ime should be given to both the tasks irresp of the keybucket
-    RunLoopTask task0 = getMockRunLoopTask(taskName0, ssp0);
-    RunLoopTask task1 = getMockRunLoopTask(taskName1, ssp1);
-
-    SystemConsumers consumerMultiplexer = mock(SystemConsumers.class);
-    when(consumerMultiplexer.choose(false)).thenReturn(envelopeEOS).thenReturn(null);
-
-    Map<TaskName, RunLoopTask> tasks = ImmutableMap.of(taskName0, task0, taskName1, task1);
-    int maxMessagesInFlight = 1;
-    RunLoop runLoop = new RunLoop(tasks, executor, consumerMultiplexer, maxMessagesInFlight, windowMs, commitMs,
-        callbackTimeoutMs, maxThrottlingDelayMs, 0, containerMetrics, () -> 0L, false, 2, null);
-    runLoop.run();
-
-    verify(task0).endOfStream(any());
-    verify(task1).endOfStream(any());
-  }
-
-  @Test
   public void testDrainWithElasticityEnabled() {
     TaskName taskName0 = new TaskName(p0.toString() + " 0");
     TaskName taskName1 = new TaskName(p0.toString() + " 1");
diff --git a/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupByPartition.java b/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupByPartition.java
index a82a9a1..e99e3f2 100644
--- a/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupByPartition.java
+++ b/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupByPartition.java
@@ -93,12 +93,12 @@
     GroupByPartition grouper = new GroupByPartition(config);
     Map<TaskName, Set<SystemStreamPartition>> result = grouper.group(ImmutableSet.of(aa0, aa1, aa2, ab1, ab2, ac0));
     Map<TaskName, Set<SystemStreamPartition>> expectedResult = ImmutableMap.<TaskName, Set<SystemStreamPartition>>builder()
-        .put(new TaskName("Partition 0_0_2"), ImmutableSet.of(new SystemStreamPartition(aa0, 0), new SystemStreamPartition(ac0, 0)))
-        .put(new TaskName("Partition 0_1_2"), ImmutableSet.of(new SystemStreamPartition(aa0, 1), new SystemStreamPartition(ac0, 1)))
-        .put(new TaskName("Partition 1_0_2"), ImmutableSet.of(new SystemStreamPartition(aa1, 0), new SystemStreamPartition(ab1, 0)))
-        .put(new TaskName("Partition 1_1_2"), ImmutableSet.of(new SystemStreamPartition(aa1, 1), new SystemStreamPartition(ab1, 1)))
-        .put(new TaskName("Partition 2_0_2"), ImmutableSet.of(new SystemStreamPartition(aa2, 0), new SystemStreamPartition(ab2, 0)))
-        .put(new TaskName("Partition 2_1_2"), ImmutableSet.of(new SystemStreamPartition(aa2, 1), new SystemStreamPartition(ab2, 1)))
+        .put(new TaskName("Partition 0 0"), ImmutableSet.of(new SystemStreamPartition(aa0, 0), new SystemStreamPartition(ac0, 0)))
+        .put(new TaskName("Partition 0 1"), ImmutableSet.of(new SystemStreamPartition(aa0, 1), new SystemStreamPartition(ac0, 1)))
+        .put(new TaskName("Partition 1 0"), ImmutableSet.of(new SystemStreamPartition(aa1, 0), new SystemStreamPartition(ab1, 0)))
+        .put(new TaskName("Partition 1 1"), ImmutableSet.of(new SystemStreamPartition(aa1, 1), new SystemStreamPartition(ab1, 1)))
+        .put(new TaskName("Partition 2 0"), ImmutableSet.of(new SystemStreamPartition(aa2, 0), new SystemStreamPartition(ab2, 0)))
+        .put(new TaskName("Partition 2 1"), ImmutableSet.of(new SystemStreamPartition(aa2, 1), new SystemStreamPartition(ab2, 1)))
         .build();
 
     assertEquals(expectedResult, result);
diff --git a/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupBySystemStreamPartition.java b/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupBySystemStreamPartition.java
index 90628b3..13b7678 100644
--- a/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupBySystemStreamPartition.java
+++ b/samza-core/src/test/java/org/apache/samza/container/grouper/stream/TestGroupBySystemStreamPartition.java
@@ -84,22 +84,14 @@
 
     Map<TaskName, Set<SystemStreamPartition>> result = grouper.group(ImmutableSet.of(aa0, aa1, aa2, ac0));
     Map<TaskName, Set<SystemStreamPartition>> expectedResult = ImmutableMap.<TaskName, Set<SystemStreamPartition>>builder()
-        .put(new TaskName(new SystemStreamPartition(aa0, 0).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa0, 0)))
-        .put(new TaskName(new SystemStreamPartition(aa0, 1).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa0, 1)))
-        .put(new TaskName(new SystemStreamPartition(aa1, 0).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa1, 0)))
-        .put(new TaskName(new SystemStreamPartition(aa1, 1).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa1, 1)))
-        .put(new TaskName(new SystemStreamPartition(aa2, 0).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa2, 0)))
-        .put(new TaskName(new SystemStreamPartition(aa2, 1).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(aa2, 1)))
-        .put(new TaskName(new SystemStreamPartition(ac0, 0).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(ac0, 0)))
-        .put(new TaskName(new SystemStreamPartition(ac0, 1).toString() + "_2"),
-            ImmutableSet.of(new SystemStreamPartition(ac0, 1)))
+        .put(new TaskName(new SystemStreamPartition(aa0, 0).toString()), ImmutableSet.of(new SystemStreamPartition(aa0, 0)))
+        .put(new TaskName(new SystemStreamPartition(aa0, 1).toString()), ImmutableSet.of(new SystemStreamPartition(aa0, 1)))
+        .put(new TaskName(new SystemStreamPartition(aa1, 0).toString()), ImmutableSet.of(new SystemStreamPartition(aa1, 0)))
+        .put(new TaskName(new SystemStreamPartition(aa1, 1).toString()), ImmutableSet.of(new SystemStreamPartition(aa1, 1)))
+        .put(new TaskName(new SystemStreamPartition(aa2, 0).toString()), ImmutableSet.of(new SystemStreamPartition(aa2, 0)))
+        .put(new TaskName(new SystemStreamPartition(aa2, 1).toString()), ImmutableSet.of(new SystemStreamPartition(aa2, 1)))
+        .put(new TaskName(new SystemStreamPartition(ac0, 0).toString()), ImmutableSet.of(new SystemStreamPartition(ac0, 0)))
+        .put(new TaskName(new SystemStreamPartition(ac0, 1).toString()), ImmutableSet.of(new SystemStreamPartition(ac0, 1)))
         .build();
 
     assertEquals(expectedResult, result);
diff --git a/samza-core/src/test/java/org/apache/samza/elasticity/util/TestElasticityUtils.java b/samza-core/src/test/java/org/apache/samza/elasticity/util/TestElasticityUtils.java
deleted file mode 100644
index dc6bfbb..0000000
--- a/samza-core/src/test/java/org/apache/samza/elasticity/util/TestElasticityUtils.java
+++ /dev/null
@@ -1,435 +0,0 @@
-/*
- * 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.samza.elasticity.util;
-
-import com.google.common.collect.ImmutableMap;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.samza.Partition;
-import org.apache.samza.checkpoint.Checkpoint;
-import org.apache.samza.checkpoint.CheckpointId;
-import org.apache.samza.checkpoint.CheckpointV2;
-import org.apache.samza.container.TaskName;
-import org.apache.samza.system.SystemAdmin;
-import org.apache.samza.system.SystemAdmins;
-import org.apache.samza.system.SystemStreamPartition;
-import org.junit.Assert;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-
-// #TODO: going to make this entire class parametrized.
-public class TestElasticityUtils {
-  private static final TaskName TASKNAME_GROUP_BY_PARTITION = new TaskName("Partition 0");
-  private static final TaskName ELASTIC_TASKNAME_GROUP_BY_PARTITION = new TaskName("Partition 0_1_2");
-  private static final TaskName TASKNAME_GROUP_BY_SSP = new TaskName("SystemStreamPartition [systemA, streamB, 0]");
-  private static final TaskName ELASTIC_TASKNAME_GROUP_BY_SSP = new TaskName("SystemStreamPartition [systemA, streamB, 0, 1]_2");
-
-  @Test
-  public void testComputeLastProcessedOffsetsFromCheckpointMap() {
-    // Setup :
-    // there is one ssp = SystemStreamPartition [systemA, streamB, partition(0)] consumed by the job
-    // Note: Partition 0_1_2 means task consumes keyBucket 1 of partition 0 and has elasticityFactor 2.
-    // Before elasticity, job has one task with name "Partition 0"
-    // with elasticity factor 2, job has 2 tasks with names "Partition 0_0_2" and "Partition 0_1_2"
-    //         Partition 0_0_2 consumes SSP[systemA, stream B, partition(0), keyBucket(0)]
-    //         Partition 0_1_2 consumes SSP[systemA, stream B, partition(0), keyBucket(1)]
-    // with elasticity factor 4, job has 4 tasks with names "Partition 0_0_4", "Partition 0_1_4", "Partition 0_2_4" and "Partition 0_3_4"
-    //         Partition 0_0_4 consumes SSP[systemA, stream B, partition(0), keyBucket(0)]
-    //         Partition 0_1_4 consumes SSP[systemA, stream B, partition(0), keyBucket(1)]
-    //         Partition 0_2_4 consumes SSP[systemA, stream B, partition(0), keyBucket(2)]
-    //         Partition 0_3_4 consumes SSP[systemA, stream B, partition(0), keyBucket(3)]
-
-    //
-    // From the definition of keyBucket computation using elasticity factor in
-    // {@link IncomingMessageEnvelope.getSystemStresamPartition(elasticityFactor) as
-    // keyBucket = (Math.abs(envelopeKeyorOffset.hashCode()) % 31) % elasticityFactor
-    // messages processed by 0_0_4 and 0_2_4 will be the same as those processed by 0_0_2
-    // messages processed by 0_1_4 and 0_3_4 will be the same as those processed by 0_1_2
-    // messages processed by 0_0_2 and 0_1_2 will be the same as those processed by Partition 0 itself
-
-    TaskName taskName = new TaskName("Partition 0_0_2");
-    Map<TaskName, Checkpoint> checkpointMap = new HashMap<>();
-    SystemStreamPartition ssp = new SystemStreamPartition("systemA", "streamB", new Partition(0));
-    SystemStreamPartition ssp0 = new SystemStreamPartition("systemA", "streamB", new Partition(0), 0);
-    SystemStreamPartition ssp2 = new SystemStreamPartition("systemA", "streamB", new Partition(0), 2);
-
-
-    SystemAdmin mockSystemAdmin = Mockito.mock(SystemAdmin.class);
-    // offsets ordering 1 < 2 < 3 < 4
-    Mockito.when(mockSystemAdmin.offsetComparator("1", "2")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("2", "1")).thenReturn(1);
-    Mockito.when(mockSystemAdmin.offsetComparator("1", "3")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("3", "1")).thenReturn(1);
-    Mockito.when(mockSystemAdmin.offsetComparator("1", "4")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("4", "1")).thenReturn(1);
-    Mockito.when(mockSystemAdmin.offsetComparator("2", "3")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("3", "2")).thenReturn(1);
-    Mockito.when(mockSystemAdmin.offsetComparator("2", "4")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("4", "2")).thenReturn(1);
-    Mockito.when(mockSystemAdmin.offsetComparator("3", "4")).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator("4", "3")).thenReturn(1);
-
-    SystemAdmins mockSystemAdmins = Mockito.mock(SystemAdmins.class);
-    Mockito.when(mockSystemAdmins.getSystemAdmin(ssp0.getSystem())).thenReturn(mockSystemAdmin);
-
-    // case 1: for task Partition 0_0_2: last deploy was with ef = 2 itself.
-    // hence "Partition 0_0_2" has the largest offset and that should be used for computing checkpoint for 0_0_2 now also
-    checkpointMap.put(new TaskName("Partition 0"), buildCheckpointV2(ssp, "1"));
-    checkpointMap.put(new TaskName("Partition 0_0_2"), buildCheckpointV2(ssp0, "4"));
-    checkpointMap.put(new TaskName("Partition 0_0_4"), buildCheckpointV2(ssp0, "2"));
-    checkpointMap.put(new TaskName("Partition 0_2_4"), buildCheckpointV2(ssp2, "3"));
-    Map<SystemStreamPartition, String> result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertEquals("4", result.get(ssp0));
-
-    // case 2: for task Partition 0_0_2: last deploy was with ef =1
-    // hence "Partition 0" has the largest offset. Computing checkpint for 0_0_2 should use this largest offset
-    checkpointMap = new HashMap<>();
-    checkpointMap.put(new TaskName("Partition 0"), buildCheckpointV2(ssp, "4"));
-    checkpointMap.put(new TaskName("Partition 0_0_2"), buildCheckpointV2(ssp0, "1"));
-    checkpointMap.put(new TaskName("Partition 0_0_4"), buildCheckpointV2(ssp0, "3"));
-    checkpointMap.put(new TaskName("Partition 0_2_4"), buildCheckpointV2(ssp2, "2"));
-
-
-    result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertEquals("4", result.get(ssp0));
-
-
-    // case 3: for task partition 0_0_2: last deploy was with ef = 4
-    // hence checkpoints of Partition 0_0_4 and Partition 0_3_4 are relevant.
-    // since messages from both end up in 0_0_2 with ef=2, need to take min of their checkpointed offsets
-
-    checkpointMap.put(new TaskName("Partition 0"), buildCheckpointV2(ssp, "1"));
-    checkpointMap.put(new TaskName("Partition 0_0_2"), buildCheckpointV2(ssp0, "2"));
-    checkpointMap.put(new TaskName("Partition 0_0_4"), buildCheckpointV2(ssp0, "3"));
-    checkpointMap.put(new TaskName("Partition 0_2_4"), buildCheckpointV2(ssp2, "4"));
-    result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertEquals("3", result.get(ssp0));
-  }
-  @Test
-  public void testComputeLastProcessedOffsetsWithEdgeCases() {
-    TaskName taskName = new TaskName("Partition 0_0_2");
-    Map<TaskName, Checkpoint> checkpointMap = new HashMap<>();
-    SystemStreamPartition ssp0 = new SystemStreamPartition("systemA", "streamB", new Partition(0), 0);
-
-    SystemAdmin mockSystemAdmin = Mockito.mock(SystemAdmin.class);
-    SystemAdmins mockSystemAdmins = Mockito.mock(SystemAdmins.class);
-    Mockito.when(mockSystemAdmins.getSystemAdmin(ssp0.getSystem())).thenReturn(mockSystemAdmin);
-
-    // case 1: empty checkpoint map
-    Map<SystemStreamPartition, String> result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertTrue("if given checkpoint map is empty, return empty last processed offsets map", result.isEmpty());
-
-    // case 2: null checkpoints given for some ancestor tasks
-    checkpointMap.put(new TaskName("Partition 0"), null);
-    result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertTrue("if given checkpoint map has null checkpoint, return empty last processed offsets map", result.isEmpty());
-
-    // case 3: null checkpoints given for some descendant tasks
-    checkpointMap.put(new TaskName("Partition 0_0_4"), null);
-    result = ElasticityUtils.computeLastProcessedOffsetsFromCheckpointMap(
-        taskName, Collections.singleton(ssp0), checkpointMap, mockSystemAdmins);
-    Assert.assertTrue("if given checkpoint map has null checkpoint, return empty last processed offsets map", result.isEmpty());
-  }
-
-  @Test
-  public void testTaskIsGroupByPartitionOrGroupBySSP() {
-    String msgPartition = "GroupByPartition task should start with Partition";
-    String msgSsp = "GroupBySystemStreamPartition task should start with SystemStreamPartition";
-
-    Assert.assertTrue(msgPartition, ElasticityUtils.isGroupByPartitionTask(TASKNAME_GROUP_BY_PARTITION));
-    Assert.assertFalse(msgPartition, ElasticityUtils.isGroupBySystemStreamPartitionTask(TASKNAME_GROUP_BY_PARTITION));
-
-    Assert.assertTrue(msgPartition, ElasticityUtils.isGroupByPartitionTask(ELASTIC_TASKNAME_GROUP_BY_PARTITION));
-    Assert.assertFalse(msgPartition, ElasticityUtils.isGroupBySystemStreamPartitionTask(
-        ELASTIC_TASKNAME_GROUP_BY_PARTITION));
-
-    Assert.assertTrue(msgSsp, ElasticityUtils.isGroupBySystemStreamPartitionTask(TASKNAME_GROUP_BY_SSP));
-    Assert.assertFalse(msgSsp, ElasticityUtils.isGroupByPartitionTask(TASKNAME_GROUP_BY_SSP));
-
-    Assert.assertTrue(msgSsp, ElasticityUtils.isGroupBySystemStreamPartitionTask(ELASTIC_TASKNAME_GROUP_BY_SSP));
-    Assert.assertFalse(msgSsp, ElasticityUtils.isGroupByPartitionTask(ELASTIC_TASKNAME_GROUP_BY_SSP));
-
-    TaskName taskName = new TaskName("FooBar");
-    Assert.assertFalse(msgPartition, ElasticityUtils.isGroupByPartitionTask(taskName));
-    Assert.assertFalse(msgSsp, ElasticityUtils.isGroupBySystemStreamPartitionTask(taskName));
-  }
-
-  @Test
-  public void testIsTaskNameElastic() {
-    Assert.assertFalse(ElasticityUtils.isTaskNameElastic(TASKNAME_GROUP_BY_SSP));
-    Assert.assertTrue(ElasticityUtils.isTaskNameElastic(ELASTIC_TASKNAME_GROUP_BY_SSP));
-    Assert.assertFalse(ElasticityUtils.isTaskNameElastic(TASKNAME_GROUP_BY_PARTITION));
-    Assert.assertTrue(ElasticityUtils.isTaskNameElastic(ELASTIC_TASKNAME_GROUP_BY_PARTITION));
-  }
-
-  @Test
-  public void testGetElasticTaskNameParts() {
-    TaskNameComponents taskNameComponents = ElasticityUtils.getTaskNameParts(TASKNAME_GROUP_BY_PARTITION);
-    Assert.assertEquals(taskNameComponents.partition, 0);
-    Assert.assertEquals(taskNameComponents.keyBucket, TaskNameComponents.DEFAULT_KEY_BUCKET);
-    Assert.assertEquals(taskNameComponents.elasticityFactor, TaskNameComponents.DEFAULT_ELASTICITY_FACTOR);
-
-    taskNameComponents = ElasticityUtils.getTaskNameParts(ELASTIC_TASKNAME_GROUP_BY_PARTITION);
-    Assert.assertEquals(taskNameComponents.partition, 0);
-    Assert.assertEquals(taskNameComponents.keyBucket, 1);
-    Assert.assertEquals(taskNameComponents.elasticityFactor, 2);
-
-    taskNameComponents = ElasticityUtils.getTaskNameParts(TASKNAME_GROUP_BY_SSP);
-    Assert.assertEquals(taskNameComponents.system, "systemA");
-    Assert.assertEquals(taskNameComponents.stream, "streamB");
-    Assert.assertEquals(taskNameComponents.partition, 0);
-    Assert.assertEquals(taskNameComponents.keyBucket, TaskNameComponents.DEFAULT_KEY_BUCKET);
-    Assert.assertEquals(taskNameComponents.elasticityFactor, TaskNameComponents.DEFAULT_ELASTICITY_FACTOR);
-
-    taskNameComponents = ElasticityUtils.getTaskNameParts(ELASTIC_TASKNAME_GROUP_BY_SSP);
-    Assert.assertEquals(taskNameComponents.system, "systemA");
-    Assert.assertEquals(taskNameComponents.stream, "streamB");
-    Assert.assertEquals(taskNameComponents.partition, 0);
-    Assert.assertEquals(taskNameComponents.keyBucket, 1);
-    Assert.assertEquals(taskNameComponents.elasticityFactor, 2);
-
-    taskNameComponents = ElasticityUtils.getTaskNameParts(new TaskName("FooBar"));
-    Assert.assertEquals(taskNameComponents.partition, TaskNameComponents.INVALID_PARTITION);
-  }
-
-  @Test
-  public void testIsOtherTaskAncestorDescendantOfCurrentTask() {
-    TaskName task0 = new TaskName("Partition 0");
-    TaskName task1 = new TaskName("Partition 1");
-    TaskName task002 = new TaskName("Partition 0_0_2");
-    TaskName task012 = new TaskName("Partition 0_1_2");
-    TaskName task004 = new TaskName("Partition 0_0_4");
-    TaskName task014 = new TaskName("Partition 0_1_4");
-    TaskName task024 = new TaskName("Partition 0_2_4");
-    TaskName task034 = new TaskName("Partition 0_3_4");
-
-    TaskName sspTask0 = new TaskName("SystemStreamPartition [systemA, streamB, 0]");
-    TaskName sspTask002 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 0]_2");
-    TaskName sspTask012 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 1]_2");
-    TaskName sspTask004 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 0]_4");
-    TaskName sspTask014 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 1]_4");
-    TaskName sspTask024 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 2]_4");
-    TaskName sspTask034 = new TaskName("SystemStreamPartition [systemA, streamB, 0, 3]_4");
-
-    // Partition 0 is ancestor of all tasks Partition 0_0_2, 0_1_2, 0_0_4, 0_1_4, 0_2_4, 0_3_4 and itself
-    // and all these tasks are descendants of Partition 0 (except itself)
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task0, task0));
-    Assert.assertFalse(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task0, task1));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task002, task0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task012, task0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task004, task0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task014, task0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task024, task0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task034, task0));
-
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task012));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task004));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task014));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task024));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task0, task034));
-
-    // Partition 0_0_2 is ancestor of tasks Partition 0_0_4 and 0_2_4 and itself
-    // these tasks are descendants of 0_0_2
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task004, task002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task024, task002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(task002, task002));
-
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task002, task004));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(task002, task024));
-
-    // "SystemStreamPartition [systemA, streamB, 0]
-    // is ancestor of all tasks "SystemStreamPartition [systemA, streamB, 0, 0]_2, [systemA, streamB, 0, 1]_2 and the rest incl itself
-    // and all these tasks are descendants of Partition 0 (except itself)
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask0, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask002, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask012, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask004, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask014, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask024, sspTask0));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask034, sspTask0));
-
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask012));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask004));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask014));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask024));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask0, sspTask034));
-
-    // SystemStreamPartition [systemA, streamB, 0, 0]_2 is ancestor of
-    // tasks SystemStreamPartition [systemA, streamB, 0, 0]_4, SystemStreamPartition [systemA, streamB, 0, 2]_4 and itself
-    // similarly, these tasks are descendants of SystemStreamPartition [systemA, streamB, 0, 0]_2
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask004, sspTask002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask024, sspTask002));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskAncestorOfCurrentTask(sspTask002, sspTask002));
-
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask002, sspTask004));
-    Assert.assertTrue(ElasticityUtils.isOtherTaskDescendantOfCurrentTask(sspTask002, sspTask024));
-  }
-
-  @Test
-  public void testGetAncestorAndDescendantCheckpoints() {
-    TaskName taskName = new TaskName("Partition 0_0_2");
-    Map<TaskName, Checkpoint> checkpointMap = new HashMap<>();
-    SystemStreamPartition ssp = new SystemStreamPartition("systemA", "streamB", new Partition(0));
-    Checkpoint ansCheckpoint1 = buildCheckpointV2(ssp, "1");
-    Checkpoint ansCheckpoint2 = buildCheckpointV2(ssp, "2");
-    Checkpoint desCheckpoint1 = buildCheckpointV2(ssp, "3");
-    Checkpoint desCheckpoint2 = buildCheckpointV2(ssp, "4");
-    Checkpoint unrelCheckpoint = buildCheckpointV2(ssp, "5");
-    Set<Checkpoint> ansCheckpointSet = new HashSet<>(Arrays.asList(ansCheckpoint1, ansCheckpoint2));
-    Set<Checkpoint> desCheckpointSet = new HashSet<>(Arrays.asList(desCheckpoint1, desCheckpoint2));
-
-    checkpointMap.put(new TaskName("Partition 0"), ansCheckpoint1);
-    checkpointMap.put(new TaskName("Partition 0_0_2"), ansCheckpoint2);
-    checkpointMap.put(new TaskName("Partition 0_0_4"), desCheckpoint1);
-    checkpointMap.put(new TaskName("Partition 0_2_4"), desCheckpoint2);
-    checkpointMap.put(new TaskName("Partition 0_1_4"), unrelCheckpoint);
-
-    Pair<Set<Checkpoint>, Map<Integer, Set<Checkpoint>>> result =
-        ElasticityUtils.getAncestorAndDescendantCheckpoints(taskName, checkpointMap);
-    Set<Checkpoint> anscestorCheckpointSet = result.getLeft();
-    Set<Checkpoint> descendantCheckpointSetForEf4 = result.getRight().get(4);
-
-    Assert.assertTrue("should contain all ancestors' checkpoints",
-        anscestorCheckpointSet.containsAll(ansCheckpointSet));
-    Assert.assertFalse("should not contain a descendant checkpoint in anscetor list",
-        anscestorCheckpointSet.contains(desCheckpoint1));
-    Assert.assertFalse("should not contain an unrelated checkpoint in ancestor list",
-        anscestorCheckpointSet.contains(unrelCheckpoint));
-
-    Assert.assertTrue("should contain all descendants' checkpoints",
-        descendantCheckpointSetForEf4.containsAll(desCheckpointSet));
-    Assert.assertFalse("should not contain a anscetor checkpoint in descendant list",
-        descendantCheckpointSetForEf4.contains(ansCheckpoint1));
-    Assert.assertFalse("should not contain an unrelated checkpoint in descendant list",
-        descendantCheckpointSetForEf4.contains(unrelCheckpoint));
-  }
-
-  @Test
-  public void testGetOffsetForSSPInCheckpoint() {
-    String offset1 = "1111";
-    String offset2 = "2222";
-    // case 1: when looking for exact ssp
-    SystemStreamPartition ssp = new SystemStreamPartition("systemA", "streamB", new Partition(0));
-    Checkpoint checkpoint1 = buildCheckpointV2(ssp, offset1);
-    Assert.assertEquals(ElasticityUtils.getOffsetForSSPInCheckpoint(checkpoint1, ssp), offset1);
-
-    // case 2: checkpoint has ssp with key bucket but looking for the full ssp (same system stream and partition but without keybucket)
-    SystemStreamPartition sspWithKB = new SystemStreamPartition("systemA", "streamB", new Partition(0), 1);
-    checkpoint1 = buildCheckpointV2(sspWithKB, offset2);
-    Assert.assertEquals(ElasticityUtils.getOffsetForSSPInCheckpoint(checkpoint1, ssp), offset2);
-
-    // case 3: try getting offset for an ssp not present in the checkpoint -> should return null
-    SystemStreamPartition ssp2 = new SystemStreamPartition("A", "B", new Partition(1));
-    Assert.assertEquals(ElasticityUtils.getOffsetForSSPInCheckpoint(checkpoint1, ssp2), null);
-  }
-
-  @Test
-  public void testGetMaxMinOffsetForSSPInCheckpointSet() {
-    String offset1 = "1111";
-    String offset2 = "2222";
-
-    SystemStreamPartition ssp = new SystemStreamPartition("systemA", "streamB", new Partition(0));
-    Checkpoint checkpoint1 = buildCheckpointV2(ssp, offset1);
-    Checkpoint checkpoint2 = buildCheckpointV2(ssp, offset2);
-    Set<Checkpoint> checkpointSet = new HashSet<>(Arrays.asList(checkpoint1, checkpoint2));
-
-    SystemAdmin mockSystemAdmin = Mockito.mock(SystemAdmin.class);
-    // offset 1 < offset2
-    Mockito.when(mockSystemAdmin.offsetComparator(offset1, offset2)).thenReturn(-1);
-    Mockito.when(mockSystemAdmin.offsetComparator(offset2, offset1)).thenReturn(1);
-
-    // case 1: when exact ssp is in checkpoint set
-    Assert.assertEquals(offset2, ElasticityUtils.getMaxOffsetForSSPInCheckpointSet(checkpointSet, ssp, mockSystemAdmin));
-    Assert.assertEquals(offset1, ElasticityUtils.getMinOffsetForSSPInCheckpointSet(checkpointSet, ssp, mockSystemAdmin));
-
-    // case 2: when looking for ssp with keyBucket 1 whereas checkpoint set only has full ssp (same system stream and partition but without keybucket)
-    SystemStreamPartition sspWithKeyBucket = new SystemStreamPartition(ssp, 1);
-    Assert.assertEquals(offset2, ElasticityUtils.getMaxOffsetForSSPInCheckpointSet(checkpointSet, sspWithKeyBucket, mockSystemAdmin));
-    Assert.assertEquals(offset1, ElasticityUtils.getMinOffsetForSSPInCheckpointSet(checkpointSet, sspWithKeyBucket, mockSystemAdmin));
-
-
-    // case 3: when ssp not in checkpoint set -> should receive null for min and max offset
-    SystemStreamPartition ssp2 = new SystemStreamPartition("A", "B", new Partition(0));
-    Assert.assertEquals(null, ElasticityUtils.getMaxOffsetForSSPInCheckpointSet(checkpointSet, ssp2, mockSystemAdmin));
-    Assert.assertEquals(null, ElasticityUtils.getMinOffsetForSSPInCheckpointSet(checkpointSet, ssp2, mockSystemAdmin));
-  }
-
-  @Test
-  public void testWasElasticityEnabled() {
-    Checkpoint checkpoint1 = buildCheckpointV2(new SystemStreamPartition("A", "B", new Partition(0)), "1");
-    Checkpoint checkpoint2 = buildCheckpointV2(new SystemStreamPartition("A", "B", new Partition(1)), "2");
-    Checkpoint checkpoint3 = buildCheckpointV2(new SystemStreamPartition("A", "B", new Partition(0), 0), "3");
-    Checkpoint checkpoint4 = buildCheckpointV2(new SystemStreamPartition("A", "B", new Partition(0), 1), "4");
-
-    // case 0: empty checkpoint map
-    Assert.assertFalse(ElasticityUtils.wasElasticityEnabled(new HashMap<>()));
-
-    // case 1: no tasks with elasticity enabled in the checkpoint map
-    Map<TaskName, Checkpoint> checkpointMap1 = new HashMap<>();
-    checkpointMap1.put(new TaskName("Partition 0"), checkpoint1);
-    checkpointMap1.put(new TaskName("Partition 2"), checkpoint2);
-    Assert.assertFalse(ElasticityUtils.wasElasticityEnabled(checkpointMap1));
-
-    // case 2: tasks with no elasticity and tasks with elasticity both present in the checkpoint map
-    Map<TaskName, Checkpoint> checkpointMap2 = new HashMap<>();
-    checkpointMap2.put(new TaskName("Partition 0"), checkpoint1);
-    checkpointMap2.put(new TaskName("Partition 2"), checkpoint2);
-    checkpointMap2.put(new TaskName("Partition 0_0_2"), checkpoint3);
-    Assert.assertTrue(ElasticityUtils.wasElasticityEnabled(checkpointMap2));
-
-    // case 3: only tasks with elasticity present in the checkpoint map
-    Map<TaskName, Checkpoint> checkpointMap3 = new HashMap<>();
-    checkpointMap3.put(new TaskName("Partition 0_0_2"), checkpoint3);
-    checkpointMap3.put(new TaskName("Partition 0_1_2"), checkpoint4);
-    Assert.assertTrue(ElasticityUtils.wasElasticityEnabled(checkpointMap3));
-
-    // case 4: repeat same checks with GroupBySSP grouper tasks
-    Map<TaskName, Checkpoint> checkpointMap4 = new HashMap<>();
-    checkpointMap4.put(new TaskName("SystemStreamPartition [A, B, 0]"), checkpoint1);
-    checkpointMap4.put(new TaskName("SystemStreamPartition [A, B, 1]"), checkpoint2);
-    Assert.assertFalse(ElasticityUtils.wasElasticityEnabled(checkpointMap4));
-    checkpointMap4.put(new TaskName("SystemStreamPartition [A, B, 0, 0]_2"), checkpoint3);
-    checkpointMap4.put(new TaskName("SystemStreamPartition [A, B, 0, 1]_2"), checkpoint4);
-    Assert.assertTrue(ElasticityUtils.wasElasticityEnabled(checkpointMap4));
-
-    // case 5: repeat same checks with AllSspToSingleTask grouper tasks - no elasticity supported for this grouper
-    Map<TaskName, Checkpoint> checkpointMap5 = new HashMap<>();
-    checkpointMap5.put(new TaskName("Task-0"), checkpoint1);
-    checkpointMap5.put(new TaskName("Task-1"), checkpoint2);
-    Assert.assertFalse(ElasticityUtils.wasElasticityEnabled(checkpointMap5));
-  }
-
-  private static CheckpointV2 buildCheckpointV2(SystemStreamPartition ssp, String offset) {
-    return new CheckpointV2(CheckpointId.create(), ImmutableMap.of(ssp, offset),
-        ImmutableMap.of("backend", ImmutableMap.of("store", "10")));
-  }
-}
diff --git a/samza-core/src/test/java/org/apache/samza/serializers/model/TestSamzaObjectMapper.java b/samza-core/src/test/java/org/apache/samza/serializers/model/TestSamzaObjectMapper.java
index 6583240..2ddecf0 100644
--- a/samza-core/src/test/java/org/apache/samza/serializers/model/TestSamzaObjectMapper.java
+++ b/samza-core/src/test/java/org/apache/samza/serializers/model/TestSamzaObjectMapper.java
@@ -340,7 +340,7 @@
     assertEquals(serializedSSPAsJson.get("partition"), deserSSPAsJson.get("partition"));
     assertEquals(serializedSSPAsJson.get("keyBucket"), deserSSPAsJson.get("-1"));
 
-    //Scenario 2: ssp serialized with new elasticMapper and deserialized by old preElastic Mapper
+    //Scenario 1: ssp serialized with new elasticMapper and deserialized by old preElastic Mapper
     SystemStreamPartition sspWithKeyBucket = new SystemStreamPartition("foo", "bar", new Partition(1), 1);
     serializedString = elasticObjectMapper.writeValueAsString(sspWithKeyBucket);
 
@@ -356,37 +356,6 @@
     assertEquals(serializedSSPAsJson.get("system"), deserSSPAsJson.get("system"));
     assertEquals(serializedSSPAsJson.get("stream"), deserSSPAsJson.get("stream"));
     assertEquals(serializedSSPAsJson.get("partition"), deserSSPAsJson.get("partition"));
-
-    // scenario 3: ssp as key serialized with preElasticMapper and deserialized by new Mapper with elasticity
-    Map<SystemStreamPartition, String> offsets = new HashMap<>();
-    String offset = "100";
-
-    String sspmapString = preElasticObjectMapper.writeValueAsString(ImmutableMap.of(ssp, offset));
-
-    TypeReference<HashMap<SystemStreamPartition, String>> typeRef
-        = new TypeReference<HashMap<SystemStreamPartition, String>>() { };
-
-    Map<SystemStreamPartition, String> deserSSPMap = elasticObjectMapper.readValue(sspmapString, typeRef);
-    SystemStreamPartition deserSSP = deserSSPMap.keySet().stream().findAny().get();
-    String deserOffset = deserSSPMap.values().stream().findFirst().get();
-    assertEquals(ssp.getSystem(), deserSSP.getSystem());
-    assertEquals(ssp.getStream(), deserSSP.getStream());
-    assertEquals(ssp.getPartition(), deserSSP.getPartition());
-    assertEquals(ssp.getKeyBucket(), deserSSP.getKeyBucket());
-    assertEquals(offset, deserOffset);
-
-    // Scenario 4: ssp key serialized with new elasticMapper and deserialized by old preElastic Mapper
-    sspmapString = elasticObjectMapper.writeValueAsString(ImmutableMap.of(sspWithKeyBucket, offset));
-
-    deserSSPMap = preElasticObjectMapper.readValue(sspmapString, typeRef);
-    deserSSP = deserSSPMap.keySet().stream().findAny().get();
-    deserOffset = deserSSPMap.values().stream().findFirst().get();
-    assertEquals(sspWithKeyBucket.getSystem(), deserSSP.getSystem());
-    assertEquals(sspWithKeyBucket.getStream(), deserSSP.getStream());
-    assertEquals(sspWithKeyBucket.getPartition(), deserSSP.getPartition());
-    // preElastic mapper does not know about KeyBucket so dont check for it
-    assertEquals(offset, deserOffset);
-
   }
 
   private JobModel deserializeFromObjectNode(ObjectNode jobModelJson) throws IOException {
@@ -440,12 +409,14 @@
   private static class OldSystemStreamPartitionKeyDeserializer extends KeyDeserializer {
     @Override
     public Object deserializeKey(String sspString, DeserializationContext ctxt) throws IOException {
-      String[] parts = sspString.split("\\.");
-      if (parts.length < 3 || parts.length > 4) {
-        throw new IllegalArgumentException("System stream partition expected in format 'system.stream.partition' "
-            + "or 'system.stream.partition.keyBucket");
+      int idx = sspString.indexOf('.');
+      int lastIdx = sspString.lastIndexOf('.');
+      if (idx < 0 || lastIdx < 0) {
+        throw new IllegalArgumentException("System stream partition expected in format 'system.stream.partition");
       }
-      return new SystemStreamPartition(new SystemStream(parts[0], parts[1]), new Partition(Integer.parseInt(parts[2])));
+      return new SystemStreamPartition(
+          new SystemStream(sspString.substring(0, idx), sspString.substring(idx + 1, lastIdx)),
+          new Partition(Integer.parseInt(sspString.substring(lastIdx + 1))));
     }
   }
   public static ObjectMapper getOldDeserForSSpKeyObjectMapper() {
@@ -483,27 +454,6 @@
     }
   }
 
-  private static class PreElasticitySystemStreamPartitionKeySerializer extends JsonSerializer<SystemStreamPartition> {
-    @Override
-    public void serialize(SystemStreamPartition ssp, JsonGenerator jgen, SerializerProvider provider) throws IOException {
-      String sspString = ssp.getSystem() + "." + ssp.getStream() + "."
-          + String.valueOf(ssp.getPartition().getPartitionId());
-      jgen.writeFieldName(sspString);
-    }
-  }
-
-  private static class PreElasticitySystemStreamPartitionKeyDeserializer extends KeyDeserializer {
-    @Override
-    public Object deserializeKey(String sspString, DeserializationContext ctxt) throws IOException {
-      String[] parts = sspString.split("\\.");
-      if (parts.length < 3) {
-        throw new IllegalArgumentException("System stream partition expected in format 'system.stream.partition");
-      }
-      return new SystemStreamPartition(
-          new SystemStream(parts[0], parts[1]), new Partition(Integer.parseInt(parts[2])));
-    }
-  }
-
   private static final ObjectMapper OBJECT_MAPPER = getPreEleasticObjectMapper();
   public static ObjectMapper getPreEleasticObjectMapper() {
     ObjectMapper mapper = new ObjectMapper();
@@ -514,13 +464,13 @@
     // Setup custom serdes for simple data types.
     module.addSerializer(Partition.class, new SamzaObjectMapper.PartitionSerializer());
     module.addSerializer(SystemStreamPartition.class, new PreElasticitySystemStreamPartitionSerializer());
-    module.addKeySerializer(SystemStreamPartition.class, new PreElasticitySystemStreamPartitionKeySerializer());
+    module.addKeySerializer(SystemStreamPartition.class, new SamzaObjectMapper.SystemStreamPartitionKeySerializer());
     module.addSerializer(TaskName.class, new SamzaObjectMapper.TaskNameSerializer());
     module.addSerializer(TaskMode.class, new SamzaObjectMapper.TaskModeSerializer());
     module.addDeserializer(TaskName.class, new SamzaObjectMapper.TaskNameDeserializer());
     module.addDeserializer(Partition.class, new SamzaObjectMapper.PartitionDeserializer());
     module.addDeserializer(SystemStreamPartition.class, new PreElasticitySystemStreamPartitionDeserializer());
-    module.addKeyDeserializer(SystemStreamPartition.class, new PreElasticitySystemStreamPartitionKeyDeserializer());
+    module.addKeyDeserializer(SystemStreamPartition.class, new SamzaObjectMapper.SystemStreamPartitionKeyDeserializer());
     module.addDeserializer(Config.class, new SamzaObjectMapper.ConfigDeserializer());
     module.addDeserializer(TaskMode.class, new SamzaObjectMapper.TaskModeDeserializer());
     module.addSerializer(CheckpointId.class, new SamzaObjectMapper.CheckpointIdSerializer());
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 d0039a9..c2d24a9 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
@@ -585,47 +585,6 @@
   }
 
   @Test
-  def testElasticityUpdateWithoutKeyBucket: Unit = {
-    // When elasticity is enabled, task consumes a part of the full SSP represented by SSP With KeyBucket.
-    // OffsetManager tracks the set of SSP with KeyBucket consumed by a task.
-    // However, after an IME processing is complete, OffsetManager.update is called without KeyBuket.
-    // OffsetManager has to find and udpate the last processed offset for the task correctly for its SSP with KeyBucket.
-    val taskName = new TaskName("c")
-    val systemStream = new SystemStream("test-system", "test-stream")
-    val partition = new Partition(0)
-    val systemStreamPartition = new SystemStreamPartition(systemStream, partition)
-    val systemStreamPartitionWithKeyBucket = new SystemStreamPartition(systemStreamPartition, 0);
-    val testStreamMetadata = new SystemStreamMetadata(systemStream.getStream, Map(partition -> new SystemStreamPartitionMetadata("0", "1", "2")).asJava)
-    val systemStreamMetadata = Map(systemStream -> testStreamMetadata)
-    val checkpointManager = getCheckpointManager(systemStreamPartition, taskName)
-    val startpointManagerUtil = getStartpointManagerUtil()
-    val systemAdmins = mock(classOf[SystemAdmins])
-    when(systemAdmins.getSystemAdmin("test-system")).thenReturn(getSystemAdmin)
-    val offsetManager = OffsetManager(systemStreamMetadata, new MapConfig, checkpointManager, startpointManagerUtil.getStartpointManager, systemAdmins, Map(), new OffsetManagerMetrics)
-    // register task and its input SSP with KeyBucket
-    offsetManager.register(taskName, Set(systemStreamPartitionWithKeyBucket))
-
-    offsetManager.start
-
-    // update is called with only the full SSP and no keyBucket information.
-    offsetManager.update(taskName, systemStreamPartition, "46")
-    // Get checkpoint snapshot like we do at the beginning of TaskInstance.commit()
-    val checkpoint46 = offsetManager.getLastProcessedOffsets(taskName)
-    offsetManager.update(taskName, systemStreamPartition, "47") // Offset updated before checkpoint
-    offsetManager.writeCheckpoint(taskName, new CheckpointV1(checkpoint46))
-    // OffsetManager correctly updates the lastProcessedOffset and checkpoint for the task and input SSP with KeyBucket.
-    assertEquals(Some("47"), offsetManager.getLastProcessedOffset(taskName, systemStreamPartitionWithKeyBucket))
-    assertEquals("46", offsetManager.offsetManagerMetrics.checkpointedOffsets.get(systemStreamPartitionWithKeyBucket).getValue)
-
-    // Now write the checkpoint for the latest offset
-    val checkpoint47 = offsetManager.getLastProcessedOffsets(taskName)
-    offsetManager.writeCheckpoint(taskName, new CheckpointV1(checkpoint47))
-
-    assertEquals(Some("47"), offsetManager.getLastProcessedOffset(taskName, systemStreamPartitionWithKeyBucket))
-    assertEquals("47", offsetManager.offsetManagerMetrics.checkpointedOffsets.get(systemStreamPartitionWithKeyBucket).getValue)
-  }
-
-  @Test
   def testStartpointUpdate: Unit = {
     val taskName = new TaskName("c")
     val systemStream = new SystemStream("test-system", "test-stream")
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
index 77b1ee3..08b082c 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
@@ -162,7 +162,7 @@
     verify(processesCounter).inc()
     verify(messagesActuallyProcessedCounter).inc()
 
-    // case 2: taskInstance processes the keyBucket=0 of the ssp and envelope is NOT from same keyBucket
+    // case 1: taskInstance processes the keyBucket=0 of the ssp and envelope is NOT from same keyBucket
     // taskInstance.process should throw the exception ssp is not registered.
     when(envelope.getSystemStreamPartition(2)).thenReturn(notProcessedSSPKeyBucket)
     val thrown = intercept[Exception] {
@@ -171,26 +171,6 @@
     assert(thrown.isInstanceOf[SamzaException])
     assert(thrown.getMessage.contains(notProcessedSSPKeyBucket.toString))
     assert(thrown.getMessage.contains("is not registered!"))
-
-    // case 3: taskInstance processes the keyBucket=0 of the ssp and envelope is watermark NOT from same keyBucket
-    // regular processing should happen as Watermark and end of stream should be processed by all tasks
-    val watermark = spy(IncomingMessageEnvelope.buildWatermarkEnvelope(SYSTEM_STREAM_PARTITION, 1234l))
-    when(watermark.getSystemStreamPartition(2)).thenReturn(notProcessedSSPKeyBucket)
-    this.taskInstance.process(watermark, coordinator, callbackFactory)
-    assertEquals(2, this.taskInstanceExceptionHandler.numTimesCalled) // case 1 and case 3
-    verify(this.task).processAsync(watermark, this.collector, coordinator, callback)
-    verify(processesCounter, times(3)).inc() // case 1, 2 and 3
-    verify(messagesActuallyProcessedCounter, times(2)).inc() // case 1 and 3
-
-    // case 4: taskInstance processes the keyBucket=0 of the ssp and envelope is EndOfStream NOT from same keyBucket
-    // regular processing should happen as Watermark and end of stream should be processed by all tasks
-    val endOfStream = spy(IncomingMessageEnvelope.buildEndOfStreamEnvelope(SYSTEM_STREAM_PARTITION))
-    when(endOfStream.getSystemStreamPartition(2)).thenReturn(notProcessedSSPKeyBucket)
-    this.taskInstance.process(endOfStream, coordinator, callbackFactory)
-    assertEquals(3, this.taskInstanceExceptionHandler.numTimesCalled) // case 1 and case 3 and case 4
-    verify(this.task).processAsync(endOfStream, this.collector, coordinator, callback)
-    verify(processesCounter, times(4)).inc() // case 1, 2, 3 and 4
-    verify(messagesActuallyProcessedCounter, times(3)).inc() // case 1 and 3 and 4
   }
 
   @Test
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
index bb89798..02f2e59 100644
--- a/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
@@ -19,11 +19,10 @@
 
 package org.apache.samza.serializers
 
-import com.fasterxml.jackson.databind.ObjectMapper
-
 import java.util
+
 import org.apache.samza.Partition
-import org.apache.samza.checkpoint.CheckpointV1
+import org.apache.samza.checkpoint.{CheckpointV1}
 import org.apache.samza.container.TaskName
 import org.apache.samza.system.SystemStreamPartition
 import org.junit.Assert._
@@ -50,81 +49,4 @@
     val checkpoint = checkpointSerde.fromBytes(checkpointBytes)
     assertNull(checkpoint)
   }
-
-  @Test
-  def testSSPWithKeyBucket {
-    // case 1: write and read with serde that is aware of KeyBucket within SSP
-    val serde = new CheckpointV1Serde
-    var offsets = Map[SystemStreamPartition, String]()
-    val ssp = new SystemStreamPartition("test-system", "test-stream",
-      new Partition(777), -1)
-    offsets += ssp -> "1"
-    val deserializedOffsets = serde.fromBytes(serde.toBytes(new CheckpointV1(offsets.asJava)))
-    assertEquals("1", deserializedOffsets.getOffsets.get(ssp))
-    assertEquals(1, deserializedOffsets.getOffsets.size)
-
-    // case 2: SSP was serialized by serde not aware of KeyBucket - aka did not put keyBucket into serialized form
-    val deserializedOffsets2 = serde.fromBytes(toBytesWithoutKeyBucket(new CheckpointV1(offsets.asJava)))
-    assertEquals("1", deserializedOffsets2.getOffsets.get(ssp))
-    assertEquals(1, deserializedOffsets2.getOffsets.size)
-
-    // case 3: SSP was serialized by serde aware of KeyBucket but deserialized by serde not aware of KeyBucket
-    val deserializedOffsets3 = fromBytesWithoutKeyBucket(serde.toBytes(new CheckpointV1(offsets.asJava)))
-    assertEquals("1", deserializedOffsets3.getOffsets.get(ssp))
-    assertEquals(1, deserializedOffsets3.getOffsets.size)
-
-    // case 4: SSP has keyBucket = 0 (aka not default -1) - serialize by serde NOT aware of keyBucket
-    // when serialized with serde not aware of keyBucket, the info about keyBucket is lost,
-    // we can only recover the system, stream and partition parts out during deserialization.
-    // hence after deser, we need to look for SSP without key bucket
-    val sspWithKeyBucket = new SystemStreamPartition("test-system", "test-stream",
-      new Partition(777), 0)
-    val sspWithoutKeyBucket = new SystemStreamPartition("test-system", "test-stream",
-      new Partition(777))
-    var offsets1 = Map[SystemStreamPartition, String]()
-    offsets1 += sspWithKeyBucket -> "1"
-
-    val deserializedOffsets4 = serde.fromBytes(toBytesWithoutKeyBucket(new CheckpointV1(offsets1.asJava)))
-    assertEquals("1", deserializedOffsets4.getOffsets.get(sspWithoutKeyBucket))
-    assertEquals(1, deserializedOffsets4.getOffsets.size)
-
-    // case 5: SSP has KeyBucket = 0, serialized by serde aware of keyBucket
-    val deserializedOffsets5 = fromBytesWithoutKeyBucket(serde.toBytes(new CheckpointV1(offsets1.asJava)))
-    assertEquals("1", deserializedOffsets5.getOffsets.get(sspWithoutKeyBucket))
-    assertEquals(1, deserializedOffsets5.getOffsets.size)
-  }
-
-  private  def fromBytesWithoutKeyBucket(bytes: Array[Byte]): CheckpointV1 = {
-    val jsonMapper = new ObjectMapper()
-    val jMap = jsonMapper.readValue(bytes, classOf[util.HashMap[String, util.HashMap[String, String]]])
-
-    def deserializeJSONMap(sspInfo:util.HashMap[String, String]) = {
-      val system = sspInfo.get("system")
-      val stream = sspInfo.get("stream")
-      val partition = sspInfo.get("partition")
-      val offset = sspInfo.get("offset")
-      new SystemStreamPartition(system, stream, new Partition(partition.toInt)) -> offset
-    }
-    val cpMap = jMap.values.asScala.map(deserializeJSONMap).toMap
-    new CheckpointV1(cpMap.asJava)
-  }
-
-  private   def toBytesWithoutKeyBucket(checkpoint: CheckpointV1): Array[Byte] = {
-    val jsonMapper = new ObjectMapper()
-    val offsets = checkpoint.getOffsets
-    val asMap = new util.HashMap[String, util.HashMap[String, String]](offsets.size())
-
-    offsets.asScala.foreach {
-      case (ssp, offset) =>
-        val jMap = new util.HashMap[String, String](4)
-        jMap.put("system", ssp.getSystemStream.getSystem)
-        jMap.put("stream", ssp.getSystemStream.getStream)
-        jMap.put("partition", ssp.getPartition.getPartitionId.toString)
-        jMap.put("offset", offset)
-
-        asMap.put(ssp.toString, jMap)
-    }
-
-    jsonMapper.writeValueAsBytes(asMap)
-  }
 }
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
index cc97a70..7793b56 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
@@ -142,7 +142,16 @@
 
     info(s"Reading checkpoint for taskName $taskName")
 
-    populateTaskNamesToCheckpointsMap()
+    if (taskNamesToCheckpoints == null) {
+      info("Reading checkpoints for the first time")
+      taskNamesToCheckpoints = readCheckpoints()
+      if (stopConsumerAfterFirstRead) {
+        info("Stopping system consumer")
+        systemConsumer.stop()
+      }
+    } else if (!stopConsumerAfterFirstRead) {
+      taskNamesToCheckpoints ++= readCheckpoints()
+    }
 
     val checkpoint: Checkpoint = taskNamesToCheckpoints.getOrElse(taskName, null)
 
@@ -151,14 +160,6 @@
   }
 
   /**
-   * @inheritdoc
-   */
-  override def readAllCheckpoints(): util.Map[TaskName, Checkpoint] = {
-    populateTaskNamesToCheckpointsMap()
-    scala.collection.JavaConverters.mapAsJavaMapConverter(taskNamesToCheckpoints).asJava
-  }
-
-  /**
     * @inheritdoc
     */
   override def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
@@ -409,17 +410,4 @@
       throw new IllegalArgumentException("Unknown checkpoint key type: " + checkpointKey.getType)
     }
   }
-
-  private def populateTaskNamesToCheckpointsMap() = {
-    if (taskNamesToCheckpoints == null) {
-      info("Reading checkpoints for the first time")
-      taskNamesToCheckpoints = readCheckpoints()
-      if (stopConsumerAfterFirstRead) {
-        info("Stopping system consumer")
-        systemConsumer.stop()
-      }
-    } else if (!stopConsumerAfterFirstRead) {
-      taskNamesToCheckpoints ++= readCheckpoints()
-    }
-  }
 }
diff --git a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.java b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.java
index d0e4586..fe9bfb1 100644
--- a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.java
+++ b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.java
@@ -386,25 +386,6 @@
   }
 
   @Test
-  public void testReadAllCheckpoints() throws InterruptedException {
-    Config config = config(ImmutableMap.of(TaskConfig.CHECKPOINT_READ_VERSIONS, "1,2"));
-    setupSystemFactory(config);
-    CheckpointV2 checkpointV2ForTask0 = buildCheckpointV2(INPUT_SSP0, "0");
-    CheckpointV2 checkpointV2ForTask1 = buildCheckpointV2(INPUT_SSP0, "1");
-    List<IncomingMessageEnvelope> checkpointEnvelopes =
-        ImmutableList.of(
-            newCheckpointV2Envelope(TASK0, checkpointV2ForTask0, "0"),
-            newCheckpointV2Envelope(TASK1, checkpointV2ForTask1, "1")
-            );
-    setupConsumer(checkpointEnvelopes);
-    Map<TaskName, Checkpoint> checkpointMap = ImmutableMap.of(TASK0, checkpointV2ForTask0, TASK1, checkpointV2ForTask1);
-    KafkaCheckpointManager kafkaCheckpointManager = buildKafkaCheckpointManager(true, config);
-    kafkaCheckpointManager.register(TASK0);
-    Map<TaskName, Checkpoint> readCheckpoints = kafkaCheckpointManager.readAllCheckpoints();
-    assertEquals(checkpointMap, readCheckpoints);
-  }
-
-  @Test
   public void testWriteCheckpointV1() {
     setupSystemFactory(config());
     KafkaCheckpointManager kafkaCheckpointManager = buildKafkaCheckpointManager(true, config());