QuorumJournalManager implements JournalManager { @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()); } } if (mostRecentSegmentTxId != Long.MIN_VALUE) { recoverUnclosedSegment(mostRecentSegmentTxId); } isActiveWriter = true; } QuorumJournalManager(Configuration conf,
      URI uri, NamespaceInfo nsInfo);   QuorumJournalManager(Configuration conf,
      URI uri, NamespaceInfo nsInfo,
      AsyncLogger.Factory loggerFactory); static void checkJournalId(String jid); @Override void format(NamespaceInfo nsInfo); @Override boolean hasSomeData(); @Override EditLogOutputStream startLogSegment(long txId, int layoutVersion); @Override void finalizeLogSegment(long firstTxId, long lastTxId); @Override void setOutputBufferCapacity(int size); @Override void purgeLogsOlderThan(long minTxIdToKeep); @Override void recoverUnfinalizedSegments(); @Override void close(); @Override void selectInputStreams(Collection<EditLogInputStream> streams,
      long fromTxnId, boolean inProgressOk); @Override String toString(); @Override void doPreUpgrade(); @Override void doUpgrade(Storage storage); @Override void doFinalize(); @Override boolean canRollBack(StorageInfo storage, StorageInfo prevStorage,
      int targetLayoutVersion); @Override void doRollback(); @Override long getJournalCTime(); @Override void discardSegments(long startTxId);  }
@Test public void testChangeWritersLogsInSync() throws Exception { writeSegment(cluster, qjm, 1, 3, false); QJMTestUtil.assertExistsInQuorum(cluster, NNStorage.getInProgressEditsFileName(1)); qjm = closeLater(new QuorumJournalManager( conf, cluster.getQuorumJournalURI(JID), FAKE_NSINFO)); qjm.recoverUnfinalizedSegments(); checkRecovery(cluster, 1, 3); }