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 .