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();
  }
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.
Or if the file is corrupted we want to update the right offsets to the main offset file .