cluster-bootstrap-api/src/main/java/com/streamsets/pipeline/SdcClusterOffsetHelper.java
private void writeOffsetsToMainOffsetFile(Map<Integer, Long> partitionToOffsetMap) throws IOException {
    LOG.info("Saving the following offset {} to {}", partitionToOffsetMap, checkPointFilePath);

    //Creating a marker file (overwriting if it already exists) to mark that we are going to write offsets out the offsets to the main offset file.
    try(OutputStream os = fs.create(checkPointMarkerFilePath, true)) {
      //NOOP
    }
    //If the both above passes and writing fails or leaves corrupted file we will have the back file
    try (OutputStream os = fs.create(checkPointFilePath, true)) {
      OBJECT_MAPPER.writeValue(os, new ClusterSourceOffsetJson(serializeKafkaPartitionOffset(partitionToOffsetMap), SDC_STREAMING_OFFSET_VERSION));
    }

    //If this fails we are still good, as we will start from the backup offset file. (Not optimal, but deterministic)
    boolean deleted = fs.delete(checkPointMarkerFilePath, false);
    LOG.warn("Status {} for Deleting Marker File {}", deleted, checkPointMarkerFilePath);

    //If the write fails we don't want to touch the timestamp and will error out so not doing this in finally
    lastOffsetStoredTime = System.currentTimeMillis();
  }
private void writeOffsetsToMainOffsetFile(Map<Integer, Long> partitionToOffsetMap) throws IOException {
    LOG.info("Saving the following offset {} to {}", partitionToOffsetMap, checkPointFilePath);

    //Creating a marker file (overwriting if it already exists) to mark that we are going to write offsets out the offsets to the main offset file.
    try(OutputStream os = fs.create(checkPointMarkerFilePath, true)) {
      //NOOP
    }
    //If the both above passes and writing fails or leaves corrupted file we will have the back file
    try (OutputStream os = fs.create(checkPointFilePath, true)) {
      OBJECT_MAPPER.writeValue(os, new ClusterSourceOffsetJson(serializeKafkaPartitionOffset(partitionToOffsetMap), SDC_STREAMING_OFFSET_VERSION));
    }

    //If this fails we are still good, as we will start from the backup offset file. (Not optimal, but deterministic)
    boolean deleted = fs.delete(checkPointMarkerFilePath, false);
    LOG.warn("Status {} for Deleting Marker File {}", deleted, checkPointMarkerFilePath);

    //If the write fails we don't want to touch the timestamp and will error out so not doing this in finally
    lastOffsetStoredTime = System.currentTimeMillis();
  }
Or if the file is corrupted, we want to update the right offsets to the main offset file.