import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
/**
 * Fence any previous writers, and obtain a unique epoch number
 * for write-access to the journal nodes.
 *
 * @return the new, unique epoch number
 */
Map<AsyncLogger, NewEpochResponseProto> createNewUniqueEpoch()
    throws IOException {
  Preconditions.checkState(!loggers.isEpochEstablished(),
      "epoch already created");
  
  Map<AsyncLogger, GetJournalStateResponseProto> lastPromises =
    loggers.waitForWriteQuorum(loggers.getJournalState(),
        getJournalStateTimeoutMs, "getJournalState()");
  
  long maxPromised = Long.MIN_VALUE;
  for (GetJournalStateResponseProto resp : lastPromises.values()) {
    maxPromised = Math.max(maxPromised, resp.getLastPromisedEpoch());
  }
  assert maxPromised >= 0;
  
  long myEpoch = maxPromised + 1;
  Map<AsyncLogger, NewEpochResponseProto> resps =
      loggers.waitForWriteQuorum(loggers.newEpoch(nsInfo, myEpoch),
          newEpochTimeoutMs, "newEpoch(" + myEpoch + ")");
      
  loggers.setEpoch(myEpoch);
  return resps;
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
public QuorumCall<AsyncLogger,NewEpochResponseProto> newEpoch(
    NamespaceInfo nsInfo,
    long epoch) {
  Map<AsyncLogger, ListenableFuture<NewEpochResponseProto>> calls =
      Maps.newHashMap();
  for (AsyncLogger logger : loggers) {
    calls.put(logger, logger.newEpoch(epoch));
  }
  return QuorumCall.create(calls);    
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<NewEpochResponseProto> newEpoch(
    final long epoch) {
  return singleThreadExecutor.submit(new Callable<NewEpochResponseProto>() {
    @Override
    public NewEpochResponseProto call() throws IOException {
      return getProxy().newEpoch(journalId, nsInfo, epoch);
    }
  });
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Override
public void recoverUnfinalizedSegments() throws IOException {
  Preconditions.checkState(!isActiveWriter, "already active writer");
  
  LOG.info("Starting recovery process for unclosed journal segments...");
  Map<AsyncLogger, NewEpochResponseProto> resps = createNewUniqueEpoch();
  LOG.info("Successfully started new epoch " + loggers.getEpoch());

  if (LOG.isDebugEnabled()) {
    LOG.debug("newEpoch(" + loggers.getEpoch() + ") responses:\n" +
      QuorumCall.mapToString(resps));
  }
  
  long mostRecentSegmentTxId = Long.MIN_VALUE;
  for (NewEpochResponseProto r : resps.values()) {
    if (r.hasLastSegmentTxId()) {
      mostRecentSegmentTxId = Math.max(mostRecentSegmentTxId,
          r.getLastSegmentTxId());
    }
  }
  
  // On a completely fresh system, none of the journals have any
  // segments, so there's nothing to recover.
  if (mostRecentSegmentTxId != Long.MIN_VALUE) {
    recoverUnclosedSegment(mostRecentSegmentTxId);
  }
  isActiveWriter = true;
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
/**
 * Try to create a new epoch for this journal.
 * @param nsInfo the namespace, which is verified for consistency or used to
 * format, if the Journal has not yet been written to.
 * @param epoch the epoch to start
 * @return the status information necessary to begin recovery
 * @throws IOException if the node has already made a promise to another
 * writer with a higher epoch number, if the namespace is inconsistent,
 * or if a disk error occurs.
 */
synchronized NewEpochResponseProto newEpoch(
    NamespaceInfo nsInfo, long epoch) throws IOException {

  checkFormatted();
  storage.checkConsistentNamespace(nsInfo);

  // Check that the new epoch being proposed is in fact newer than
  // any other that we've promised. 
  if (epoch <= getLastPromisedEpoch()) {
    throw new IOException("Proposed epoch " + epoch + " <= last promise " +
        getLastPromisedEpoch());
  }
  
  updateLastPromisedEpoch(epoch);
  abortCurSegment();
  
  NewEpochResponseProto.Builder builder =
      NewEpochResponseProto.newBuilder();

  EditLogFile latestFile = scanStorageForLatestEdits();

  if (latestFile != null) {
    builder.setLastSegmentTxId(latestFile.getFirstTxId());
  }
  
  return builder.build();
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Override
public NewEpochResponseProto newEpoch(RpcController controller,
    NewEpochRequestProto request) throws ServiceException {
  try {
    return impl.newEpoch(
        request.getJid().getIdentifier(),
        PBHelper.convert(request.getNsInfo()),
        request.getEpoch());
  } catch (IOException ioe) {
    throw new ServiceException(ioe);
  }
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Override
public NewEpochResponseProto newEpoch(String jid, NamespaceInfo nsInfo,
    long epoch) throws IOException {
  try {
    NewEpochRequestProto req = NewEpochRequestProto.newBuilder()
      .setJid(convertJournalId(jid))
      .setNsInfo(PBHelper.convert(nsInfo))
      .setEpoch(epoch)
      .build();
    return rpcProxy.newEpoch(NULL_CONTROLLER, req);
  } catch (ServiceException e) {
    throw ProtobufHelper.getRemoteException(e);
  }
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  spyLoggers = ImmutableList.of(
      mockLogger(),
      mockLogger(),
      mockLogger());

  qjm = new QuorumJournalManager(conf, new URI("qjournal://host/jid"), FAKE_NSINFO) {
    @Override
    protected List<AsyncLogger> createLoggers(AsyncLogger.Factory factory) {
      return spyLoggers;
    }
  };

  for (AsyncLogger logger : spyLoggers) {
    futureReturns(GetJournalStateResponseProto.newBuilder()
        .setLastPromisedEpoch(0)
        .setHttpPort(-1)
        .build())
      .when(logger).getJournalState();
    
    futureReturns(
        NewEpochResponseProto.newBuilder().build()
        ).when(logger).newEpoch(Mockito.anyLong());
    
    futureReturns(null).when(logger).format(Mockito.<NamespaceInfo>any());
  }
  
  qjm.recoverUnfinalizedSegments();
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
@Test(timeout=100000)
public void testReturnsSegmentInfoAtEpochTransition() throws Exception {
  ch.newEpoch(1).get();
  ch.setEpoch(1);
  ch.startLogSegment(1, NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION).get();
  ch.sendEdits(1L, 1, 2, QJMTestUtil.createTxnData(1, 2)).get();
  
  // Switch to a new epoch without closing earlier segment
  NewEpochResponseProto response = ch.newEpoch(2).get();
  ch.setEpoch(2);
  assertEquals(1, response.getLastSegmentTxId());
  
  ch.finalizeLogSegment(1, 2).get();
  
  // Switch to a new epoch after just closing the earlier segment.
  response = ch.newEpoch(3).get();
  ch.setEpoch(3);
  assertEquals(1, response.getLastSegmentTxId());
  
  // Start a segment but don't write anything, check newEpoch segment info
  ch.startLogSegment(3, NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION).get();
  response = ch.newEpoch(4).get();
  ch.setEpoch(4);
  // Because the new segment is empty, it is equivalent to not having
  // started writing it. Hence, we should return the prior segment txid.
  assertEquals(1, response.getLastSegmentTxId());
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.NewEpochResponseProto; //导入依赖的package包/类
/**
 * Test that, if the writer crashes at the very beginning of a segment,
 * before any transactions are written, that the next newEpoch() call
 * returns the prior segment txid as its most recent segment.
 */
@Test (timeout = 10000)
public void testNewEpochAtBeginningOfSegment() throws Exception {
  journal.newEpoch(FAKE_NSINFO, 1);
  journal.startLogSegment(makeRI(1), 1,
      NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION);
  journal.journal(makeRI(2), 1, 1, 2, 
      QJMTestUtil.createTxnData(1, 2));
  journal.finalizeLogSegment(makeRI(3), 1, 2);
  journal.startLogSegment(makeRI(4), 3,
      NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION);
  NewEpochResponseProto resp = journal.newEpoch(FAKE_NSINFO, 2);
  assertEquals(1, resp.getLastSegmentTxId());
}