QuorumJournalManager implements JournalManager { @Override public void selectInputStreams(Collection<EditLogInputStream> streams, long fromTxnId, boolean inProgressOk) throws IOException { QuorumCall<AsyncLogger, RemoteEditLogManifest> q = loggers.getEditLogManifest(fromTxnId, inProgressOk); Map<AsyncLogger, RemoteEditLogManifest> resps = loggers.waitForWriteQuorum(q, selectInputStreamsTimeoutMs, "selectInputStreams"); LOG.debug("selectInputStream manifests:\n" + Joiner.on("\n").withKeyValueSeparator(": ").join(resps)); final PriorityQueue<EditLogInputStream> allStreams = new PriorityQueue<EditLogInputStream>(64, JournalSet.EDIT_LOG_INPUT_STREAM_COMPARATOR); for (Map.Entry<AsyncLogger, RemoteEditLogManifest> e : resps.entrySet()) { AsyncLogger logger = e.getKey(); RemoteEditLogManifest manifest = e.getValue(); for (RemoteEditLog remoteLog : manifest.getLogs()) { URL url = logger.buildURLToFetchLogs(remoteLog.getStartTxId()); EditLogInputStream elis = EditLogFileInputStream.fromUrl( connectionFactory, url, remoteLog.getStartTxId(), remoteLog.getEndTxId(), remoteLog.isInProgress()); allStreams.add(elis); } } JournalSet.chainAndMakeRedundantStreams(streams, allStreams, fromTxnId); } 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 testSelectInputStreamsMajorityDown() throws Exception { cluster.shutdown(); List<EditLogInputStream> streams = Lists.newArrayList(); try { qjm.selectInputStreams(streams, 0, false); fail("Did not throw IOE"); } catch (QuorumException ioe) { GenericTestUtils.assertExceptionContains( "Got too many exceptions", ioe); assertTrue(streams.isEmpty()); } }
@Test public void testSelectInputStreamsNotOnBoundary() throws Exception { final int txIdsPerSegment = 10; for (int txid = 1; txid <= 5 * txIdsPerSegment; txid += txIdsPerSegment) { writeSegment(cluster, qjm, txid, txIdsPerSegment, true); } File curDir = cluster.getCurrentDir(0, JID); GenericTestUtils.assertGlobEquals(curDir, "edits_.*", NNStorage.getFinalizedEditsFileName(1, 10), NNStorage.getFinalizedEditsFileName(11, 20), NNStorage.getFinalizedEditsFileName(21, 30), NNStorage.getFinalizedEditsFileName(31, 40), NNStorage.getFinalizedEditsFileName(41, 50)); ArrayList<EditLogInputStream> streams = new ArrayList<EditLogInputStream>(); qjm.selectInputStreams(streams, 25, false); verifyEdits(streams, 25, 50); }