public synchronized void startLogSegment ( RequestInfo reqInfo , long txid ) throws IOException { assert fjm != null ; checkJournalStorageFormatted ( ) ; checkRequest ( reqInfo ) ; if ( curSegment != null ) { LOG . warn ( "Client is requesting a new log segment " + txid + " though we are already writing " + curSegment + ". " + "Aborting the current segment in order to begin the new one." ) ; // The writer may have lost a connection to us and is now // re-connecting after the connection came back. // We should abort our own old segment. abortCurSegment ( ) ; } // Paranoid sanity check: we should never overwrite a finalized log file. // Additionally, if it's in-progress, it should have at most 1 transaction. // This can happen if the writer crashes exactly at the start of a segment. EditLogFile existing = fjm . getLogFile ( txid ) ; if ( existing != null ) { if ( ! existing . isInProgress ( ) ) { throw new IllegalStateException ( "Already have a finalized segment " + existing + " beginning at " + txid ) ; } // If it's in-progress, it should only contain one transaction, // because the "startLogSegment" transaction is written alone at the // start of each segment. existing . validateLog ( ) ; if ( existing . getLastTxId ( ) != existing . getFirstTxId ( ) ) { throw new IllegalStateException ( "The log file " + existing + " seems to contain valid transactions" ) ; } } long curLastWriterEpoch = lastWriterEpoch . get ( ) ; if ( curLastWriterEpoch != reqInfo . getEpoch ( ) ) { LOG . info ( "Updating lastWriterEpoch from " + curLastWriterEpoch + " to " + reqInfo . getEpoch ( ) + " for client " + Server . getRemoteIp ( ) ) ; lastWriterEpoch . set ( reqInfo . getEpoch ( ) ) ; metrics . lastWriterEpoch . set ( reqInfo . getEpoch ( ) ) ; } // The fact that we are starting a segment at this txid indicates // that any previous recovery for this same segment was aborted. // Otherwise, no writer would have started writing. So, we can // remove the record of the older segment here. purgePaxosDecision ( txid ) ; curSegment = fjm . startLogSegment ( txid ) ; curSegmentTxId = txid ; nextTxId = txid ; // the layout version has only been written // plus OP_INVALID currentSegmentWrittenBytes = 5L ; // inform the syncer service that we might have some work to do if ( journalNode != null ) { journalNode . addSyncTask ( this , curSegmentTxId ) ; } }
Start a new segment at the given txid . The previous segment must have already been finalized .