cluster-bootstrap-api/src/main/java/com/streamsets/pipeline/SdcClusterOffsetHelper.java
private Map<Integer, Long> readClusterOffsetFile(Path checkPointFilePath, int numberOfPartitions) throws IOException{
    if (!fs.exists(checkPointFilePath)) {
      throw new IOException(Utils.format("Checkpoint file path {} does not exist", checkPointFilePath));
    }
    ClusterSourceOffsetJson clusterSourceOffsetJson = OBJECT_MAPPER.readValue(
        (InputStream) fs.open(checkPointFilePath),
        ClusterSourceOffsetJson.class
    );
    String lastSourceOffset = clusterSourceOffsetJson.getOffset();
    if (!StringUtils.isEmpty(lastSourceOffset)) {
      return deserializeKafkaPartitionOffset(lastSourceOffset, numberOfPartitions);
    } else {
      throw new IOException("Partition Offset Cannot be empty");
    }
  }
private Map<Integer, Long> readClusterOffsetFile(Path checkPointFilePath, int numberOfPartitions) throws IOException{
    if (!fs.exists(checkPointFilePath)) {
      throw new IOException(Utils.format("Checkpoint file path {} does not exist", checkPointFilePath));
    }
    ClusterSourceOffsetJson clusterSourceOffsetJson = OBJECT_MAPPER.readValue(
        (InputStream) fs.open(checkPointFilePath),
        ClusterSourceOffsetJson.class
    );
    String lastSourceOffset = clusterSourceOffsetJson.getOffset();
    if (!StringUtils.isEmpty(lastSourceOffset)) {
      return deserializeKafkaPartitionOffset(lastSourceOffset, numberOfPartitions);
    } else {
      throw new IOException("Partition Offset Cannot be empty");
    }
  }
Ex: Partition Offset empty, empty map in the offset, cannot deserialize