QuorumJournalManager implements JournalManager { public void selectInputStreams(Collection<EditLogInputStream> streams, long fromTxnId, boolean inProgressOk) throws IOException { selectInputStreams(streams, fromTxnId, inProgressOk, false); } @VisibleForTesting  QuorumJournalManager(Configuration conf,
                              URI uri,
                              NamespaceInfo nsInfo);  QuorumJournalManager(Configuration conf,
      URI uri, NamespaceInfo nsInfo, String nameServiceId); @VisibleForTesting  QuorumJournalManager(Configuration conf,
                       URI uri, NamespaceInfo nsInfo,
                       AsyncLogger.Factory loggerFactory);   QuorumJournalManager(Configuration conf,
      URI uri, NamespaceInfo nsInfo, String nameServiceId,
      AsyncLogger.Factory loggerFactory); static void checkJournalId(String jid); @Override void format(NamespaceInfo nsInfo, boolean force); @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(); void selectInputStreams(Collection<EditLogInputStream> streams,
      long fromTxnId, boolean inProgressOk); @Override void selectInputStreams(Collection<EditLogInputStream> streams,
      long fromTxnId, boolean inProgressOk,
      boolean onlyDurableTxns); @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 void discardSegments(long startTxId); @Override long getJournalCTime(); static final String QJM_RPC_MAX_TXNS_KEY; static final int QJM_RPC_MAX_TXNS_DEFAULT; }
@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); }
@Test public void testSelectViaRpcWithoutDurableTransactions() throws Exception { setupLoggers345(); futureThrows(new IOException()).when(spies.get(1)).getJournaledEdits(1, QuorumJournalManager.QJM_RPC_MAX_TXNS_DEFAULT); List<EditLogInputStream> streams = new ArrayList<>(); qjm.selectInputStreams(streams, 1, true, false); verifyEdits(streams, 1, 5); IOUtils.closeStreams(streams.toArray(new Closeable[0])); for (AsyncLogger logger : spies) { Mockito.verify(logger, Mockito.times(1)).getJournaledEdits(1, QuorumJournalManager.QJM_RPC_MAX_TXNS_DEFAULT); } }