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");
    }
  }
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
Ex : Partition Offset empty empty map in the offset cannot deserialize