Start writing to the log segment with the given txid.
Transitions from BETWEEN_LOG_SEGMENTS state to IN_LOG_SEGMENT state.
191376
synchronized void startLogSegment(final long segmentTxId,
      boolean writeHeaderTxn) throws IOException {
    LOG.info("Starting log segment at " + segmentTxId);
    if (segmentTxId < 0) {
      throw new IOException("Bad txid: " + segmentTxId);
    }
    if (state != State.BETWEEN_LOG_SEGMENTS) {
      throw new IOException("Bad state: " + state);
    }
    if (segmentTxId <= curSegmentTxId) {
      throw new IOException("Cannot start writing to log segment "
          + segmentTxId + " when previous log segment started at "
          + curSegmentTxId);
    }
    if (segmentTxId != txid + 1) {
      throw new IOException("Cannot start log segment at txid " + segmentTxId
          + " when next expected " + (txid + 1));
    }

    numTransactions = totalTimeTransactions = numTransactionsBatchedInSync = 0;

    // TODO no need to link this back to storage anymore!
    // See HDFS-2174.
    storage.attemptRestoreRemovedStorage();
    try {
      editLogStream = journalSet.startLogSegment(segmentTxId);
    } catch (IOException ex) {
      throw new IOException("Unable to start log segment " + segmentTxId
          + ": no journals successfully started.");

    }
    curSegmentTxId = segmentTxId;
    state = State.IN_SEGMENT;
    if (writeHeaderTxn) {
      logEdit(LogSegmentOp.getInstance(FSEditLogOpCodes.OP_START_LOG_SEGMENT));
      logSync();
    }
    
    // force update of journal and image metrics
    journalSet.updateJournalMetrics();
    // If it is configured, we want to schedule an automatic edits roll
    if (timeoutRollEdits > 0) {
      FSNamesystem fsn = this.journalSet.getImage().getFSNamesystem();
      if (fsn != null) {
        // In some test cases fsn is NULL in images. Simply skip the feature.
        AutomaticEditsRoller aer = fsn.automaticEditsRoller;
        if (aer != null) {
          aer.setNextRollTime(System.currentTimeMillis() + timeoutRollEdits);
        } else {
          LOG.warn("Automatic edits roll is enabled but the roller thread "
              + "is not enabled. Should only happen in unit tests.");
        }
      } else {
        LOG.warn("FSNamesystem is NULL in FSEditLog.");
      }
    }
  }
468304
synchronized void offer(TransactionTask task)
    {
        Iv2Trace.logTransactionTaskQueueOffer(task);
        TransactionState txnState = task.getTransactionState();
        if (!m_backlog.isEmpty()) {
            /*
             * This branch happens during regular execution when a multi-part is in progress.
             * The first task for the multi-part is the head of the queue, and all the single parts
             * are being queued behind it. The txnid check catches tasks that are part of the multi-part
             * and immediately queues them for execution. If any multi-part txn with smaller txnId shows up,
             * it must from repair process, just let it through.
             */
            if (txnState.isSinglePartition() ){
                m_backlog.addLast(task);
                return;
            }

            //It is possible a RO MP read with higher TxnId could be executed before a RO MP reader with lower TxnId
            //so do not offer them to the site task queue in the same time, place it in the backlog instead. However,
            //if it is an MP Write with a lower TxnId than the TxnId at the head of the backlog it could be a repair
            //task so put the MP Write task into the Scoreboard or the SiteTaskQueue
            TransactionTask headTask = m_backlog.getFirst();
            if (txnState.isReadOnly() && headTask.getTransactionState().isReadOnly() ?
                    TxnEgo.getSequence(task.getTxnId()) != TxnEgo.getSequence(headTask.getTxnId()) :
                    TxnEgo.getSequence(task.getTxnId()) > TxnEgo.getSequence(headTask.getTxnId())) {
                m_backlog.addLast(task);
            } else if (task.needCoordination() && m_scoreboardEnabled) {
                /*
                 * This branch coordinates FragmentTask or CompletedTransactionTask,
                 * holds the tasks until all the sites on the node receive the task.
                 * Task with newer spHandle will
                 */
                coordinatedTaskQueueOffer(task);
            } else {
                taskQueueOffer(task);
            }
        } else {
            /*
             * Base case nothing queued nothing in progress
             * If the task is a multipart then put an entry in the backlog which
             * will act as a barrier for single parts, queuing them for execution after the
             * multipart
             */
            if (!txnState.isSinglePartition()) {
                m_backlog.addLast(task);
            }
            /*
             * This branch coordinates FragmentTask or CompletedTransactionTask,
             * holds the tasks until all the sites on the node receive the task.
             * Task with newer spHandle will
             */
            if (task.needCoordination() && m_scoreboardEnabled) {
                coordinatedTaskQueueOffer(task);
            } else {
                taskQueueOffer(task);
            }
        }
    }
1315937
public void purgeLogsOlderThan(final long minTxIdToKeep) {
    synchronized (this) {
      // synchronized to prevent findbugs warning about inconsistent
      // synchronization. This will be JIT-ed out if asserts are
      // off.
      assert curSegmentTxId == HdfsConstants.INVALID_TXID || // on format this
                                                             // is no-op
          minTxIdToKeep <= curSegmentTxId : "cannot purge logs older than txid "
          + minTxIdToKeep + " when current segment starts at " + curSegmentTxId;
      try {
        journalSet.purgeLogsOlderThan(minTxIdToKeep);
      } catch (IOException ex) {
        // All journals have failed, it will be handled in logSync.
      }

    }
  }
905720
public synchronized void finalizeLogSegment(RequestInfo reqInfo, long startTxId,
      long endTxId) throws IOException {
    checkJournalStorageFormatted();
    checkRequest(reqInfo);
    
    boolean needsValidation = true;

    // Finalizing the log that the writer was just writing.
    if (startTxId == curSegmentTxId) {
      if (curSegment != null) {
        curSegment.close();
        curSegment = null;
        curSegmentTxId = HdfsConstants.INVALID_TXID;
        currentSegmentWrittenBytes = 0L;
      }
      
      checkSync(nextTxId == endTxId + 1,
          "Trying to finalize in-progress log segment %s to end at " +
          "txid %s but only written up to txid %s",
          startTxId, endTxId, nextTxId - 1);
      // No need to validate the edit log if the client is finalizing
      // the log segment that it was just writing to.
      needsValidation = false;
    }
    
    FileJournalManager.EditLogFile elf = fjm.getLogFile(startTxId);
    if (elf == null) {
      throw new JournalOutOfSyncException("No log file to finalize at " +
          "transaction ID " + startTxId);
    }

    if (elf.isInProgress()) {
      if (needsValidation) {
        LOG.info("Validating log segment " + elf.getFile() + " about to be " +
            "finalized");
        elf.validateLog();
  
        checkSync(elf.getLastTxId() == endTxId,
            "Trying to finalize in-progress log segment %s to end at " +
            "txid %s but log %s on disk only contains up to txid %s",
            startTxId, endTxId, elf.getFile(), elf.getLastTxId());
      }
      fjm.finalizeLogSegment(startTxId, endTxId);
    } else {
      Preconditions.checkArgument(endTxId == elf.getLastTxId(),
          "Trying to re-finalize already finalized log " +
              elf + " with different endTxId " + endTxId);
    }

    // Once logs are finalized, a different length will never be decided.
    // During recovery, we treat a finalized segment the same as an accepted
    // recovery. Thus, we no longer need to keep track of the previously-
    // accepted decision. The existence of the finalized log segment is enough.
    purgePaxosDecision(elf.getFirstTxId());
  }
1188388
public void write(WARCRecord record) throws IOException {
        if (bytesWritten > maxSegmentSize) {
            dataStream.close();
            createSegment();
        }
        record.write(dataStream);
    }
760712
public void start() throws Exception {
        started = true;
        setLogFileDir(logFileDir);
        LOGGER.debug("Initiating transaction manager recovery");
        recovered = new HashMap<>();

        logger.open(null);

        ReplayListener replayListener = new GeronimoReplayListener(xidFactory, recovered);
        logger.replayActiveTx(replayListener);

        LOGGER.debug("In doubt transactions recovered from log");
    }
779138
private void writeStartSequence(IXADataRecorder dataRecorder)
			throws IOException, InterruptedException {

		LogRecordWriter writer = new LogRecordWriter();
		writer.writeLong(dataRecorder.getXADataRecorderId());
		dataRecorder.writeRollbackData(writer.toByteArray());
	}
20031
@Override
	public void onTxRollback(Transaction tx) {
		if (!tx.isReadOnly() && enableLogging) {
			rollback(tx);
			LogSeqNum lsn = new RollbackRecord(txNum).writeToLog();
			VanillaDb.logMgr().flush(lsn);
		}
	}
361666
void deliverUndeliveredTxns() throws IOException {
    LOG.debug("Delivering all the undelivered txns after {}",
              lastDeliveredZxid);
    Zxid startZxid =
      new Zxid(lastDeliveredZxid.getEpoch(), lastDeliveredZxid.getXid() + 1);
    Log log = persistence.getLog();
    try (Log.LogIterator iter = log.getIterator(startZxid)) {
      while (iter.hasNext()) {
        Transaction txn = iter.next();
        if (txn.getType() == ProposalType.USER_REQUEST_VALUE) {
          // Only delivers REQUEST proposal, ignores COP.
          stateMachine.deliver(txn.getZxid(), txn.getBody(), null, null);
        }
        lastDeliveredZxid = txn.getZxid();
      }
    }
  }
160008
private Transaction createTransaction(long writePointer, TransactionType type) {
    // For holding the first in progress short transaction Id (with timeout >= 0).
    long firstShortTx = Transaction.NO_TX_IN_PROGRESS;
    LongArrayList inProgressIds = new LongArrayList(inProgress.size());
    for (Map.Entry<Long, InProgressTx> entry : inProgress.entrySet()) {
      long txId = entry.getKey();
      inProgressIds.add(txId);
      // add any checkpointed write pointers to the in-progress list
      LongArrayList childIds = entry.getValue().getCheckpointWritePointers();
      if (childIds != null) {
        for (int i = 0; i < childIds.size(); i++) {
          inProgressIds.add(childIds.get(i));
        }
      }
      if (firstShortTx == Transaction.NO_TX_IN_PROGRESS && !entry.getValue().isLongRunning()) {
        firstShortTx = txId;
      }
    }

    return new Transaction(readPointer, writePointer, invalidArray, inProgressIds.toLongArray(), firstShortTx, type);
  }
638037
private void checkProgress() throws IOException {
    if (now() - mostRecentlyReadTransactionTime > nothingReadCheckTimeout) {
      // check the journal manager
      detectJournalManager();
      
      // we haven't read anything in a while
      if (segmentExists(mostRecentlyReadTransactionTxId + 1) &&
          mostRecentlyReadTransactionTxId != -1) {
        // unclean shutdown happened, we need to switch to next segment
        currentSegmentTxId = mostRecentlyReadTransactionTxId + 1;
        // indicate that we will reopen new stream
        currentEditLogInputStreamPosition = -1;
        
        LOG.warn("Segment - switching to new log segment with txid: "
            + currentSegmentTxId + ". Previous segment was not closed cleanly.");
      } else {
        LOG.warn("Segment - will reopen segment with txid: "
            + currentSegmentTxId + " at position: "
            + currentEditLogInputStreamPosition);
      }
      // in both cases we want to reopen the stream
      currentEditLogInputStream = null;
    }
  }
184585
void writeTransitions(
        int size,
        ObjectOutput out
    ) throws IOException {

        SPX.writeTransitions(this.transitions, size, out);

    }
628607
void prepareForWrite(long xaDataRecorderId) throws IOException, InterruptedException {
      this.dataLogger.reopen(AccessMode.WRITE);
      this.writeStartSequence(xaDataRecorderId);
   }
175051
def start_transaction(self, read_concern=None, write_concern=None,
                          read_preference=None):
        
        self._check_ended()

        if self._in_transaction:
            raise InvalidOperation("Transaction already in progress")

        read_concern = self._inherit_option("read_concern", read_concern)
        write_concern = self._inherit_option("write_concern", write_concern)
        read_preference = self._inherit_option(
            "read_preference", read_preference)

        self._transaction.opts = TransactionOptions(
            read_concern, write_concern, read_preference)
        self._transaction.reset()
        self._transaction.state = _TxnState.STARTING
        self._start_retryable_write()
        return _TransactionContext(self)
393897
private void handleLogSealed(Segment segment) {
        sealedSegmentQueue.add(segment);
        retransmitPool.execute(() -> {
            Retry.indefinitelyWithExpBackoff(config.getInitalBackoffMillis(), config.getBackoffMultiple(),
                                             config.getMaxBackoffMillis(),
                                             t -> log.error("Encountered excemption when handeling a sealed segment: ", t))
                 .run(() -> {
                     /*
                      * Using writeSealLock prevents concurrent segmentSealedCallback for different segments
                      * from being invoked concurrently, or concurrently with write.
                      * 
                      * By calling flush while the write lock is held we can ensure that any inflight
                      * entries that will succeed in being written to a new segment are written and any
                      * segmentSealedCallbacks that will be called happen before the next write is invoked.
                      */
                     synchronized (writeSealLock) {
                         Segment toSeal = sealedSegmentQueue.poll();
                         log.info("Sealing segment {} ", toSeal);
                         while (toSeal != null) {
                             resend(selector.refreshSegmentEventWritersUponSealed(toSeal, segmentSealedCallBack));
                             /* In the case of segments merging Flush ensures there can't be anything left
                              * inflight that will need to be resent to the new segment when the write lock
                              * is released. (To preserve order)
                              */
                             for (SegmentOutputStream writer : selector.getWriters()) {
                                 try {
                                     writer.write(PendingEvent.withoutHeader(null, ByteBufferUtils.EMPTY, null));
                                     writer.flush();
                                 } catch (SegmentSealedException e) {
                                     // Segment sealed exception observed during a flush. Re-run flush on all the
                                     // available writers.
                                     log.info("Flush on segment {} failed due to {}, it will be retried.", writer.getSegmentName(), e.getMessage());
                                 }
                             }
                             toSeal = sealedSegmentQueue.poll();
                             log.info("Sealing another segment {} ", toSeal);
                         }
                     }
                     return null;
                 });
        });
    }
573818
private void transferEntry(long index, Entry entry, Segment compactSegment) {
    compactSegment.append(entry);

    // If the entry was released in the prior segment, mark it as released in the compact segment.
    if (!segment.isLive(index)) {
      compactSegment.release(index);
    }
  }
1287769
synchronized void endCurrentLogSegment(boolean writeEndTxn)
      throws IOException {
    LOG.info("Ending log segment " + curSegmentTxId);
    if (state != State.IN_SEGMENT) {
      throw new IllegalStateException("Bad state: " + state);
    }
    waitForSyncToFinish();
    if (writeEndTxn) {
      logEdit(LogSegmentOp.getInstance(FSEditLogOpCodes.OP_END_LOG_SEGMENT));
    }
    logSyncAll();
    printStatistics(true);
    final long lastTxId = getLastWrittenTxId();

    try {
      journalSet.finalizeLogSegment(curSegmentTxId, lastTxId);
      editLogStream = null;
    } catch (IOException e) {
      // All journals have failed, it will be handled in logSync.
      FSNamesystem.LOG.info("Cannot finalize log segment: " + e.toString());
    }
    state = State.BETWEEN_LOG_SEGMENTS;
  }