@Test public void catchUpInSteps() throws Exception { CountingDummyFileSystemMaster countingMaster = new CountingDummyFileSystemMaster(); mFollowerJournalSystem.createJournal(countingMaster); mFollowerJournalSystem.suspend(); final int entryBatchCount = 5; try (JournalContext journalContext = mLeaderJournalSystem.createJournal(new NoopMaster()).createJournalContext()) { for (int i = 0; i < entryBatchCount; i++) { journalContext.append( alluxio.proto.journal.Journal.JournalEntry.newBuilder().setInodeLastModificationTime( File.InodeLastModificationTimeEntry.newBuilder().setId(i).build()).build()); } } Map<String, Long> backupSequences = new HashMap<>(); backupSequences.put("FileSystemMaster", (long) (entryBatchCount * 2) - 1); CatchupFuture catchupFuture = mFollowerJournalSystem.catchup(backupSequences); try (JournalContext journalContext = mLeaderJournalSystem.createJournal(new NoopMaster()).createJournalContext()) { for (int i = 0; i < entryBatchCount; i++) { journalContext.append( alluxio.proto.journal.Journal.JournalEntry.newBuilder().setInodeLastModificationTime( File.InodeLastModificationTimeEntry.newBuilder().setId(i).build()).build()); } } catchupFuture.waitTermination(); Assert.assertEquals(entryBatchCount * 2, countingMaster.getApplyCount()); mFollowerJournalSystem.catchup(backupSequences); Assert.assertEquals(entryBatchCount * 2, countingMaster.getApplyCount()); }
@Override public JournalContext createJournalContext() throws UnavailableException { AsyncJournalWriter journalWriter = mJournalWriter.get(); if (journalWriter == null) { throw new UnavailableException("Journal has been closed"); } return new MasterJournalContext(journalWriter); }
RaftJournal implements Journal { @Override public JournalContext createJournalContext() throws UnavailableException { AsyncJournalWriter journalWriter = mJournalWriter.get(); if (journalWriter == null) { throw new UnavailableException("Journal has been closed"); } return new MasterJournalContext(journalWriter); } }
RaftJournal implements Journal { @Override public JournalContext createJournalContext() throws UnavailableException { AsyncJournalWriter journalWriter = mJournalWriter.get(); if (journalWriter == null) { throw new UnavailableException("Journal has been closed"); } return new MasterJournalContext(journalWriter); } RaftJournal(Journaled stateMachine, URI location,
      AtomicReference<AsyncJournalWriter> journalWriter); }
RaftJournal implements Journal { @Override public JournalContext createJournalContext() throws UnavailableException { AsyncJournalWriter journalWriter = mJournalWriter.get(); if (journalWriter == null) { throw new UnavailableException("Journal has been closed"); } return new MasterJournalContext(journalWriter); } RaftJournal(Journaled stateMachine, URI location,
      AtomicReference<AsyncJournalWriter> journalWriter); Journaled getStateMachine(); @Override URI getLocation(); @Override JournalContext createJournalContext(); @Override void close(); }
RaftJournal implements Journal { @Override public JournalContext createJournalContext() throws UnavailableException { AsyncJournalWriter journalWriter = mJournalWriter.get(); if (journalWriter == null) { throw new UnavailableException("Journal has been closed"); } return new MasterJournalContext(journalWriter); } RaftJournal(Journaled stateMachine, URI location,
      AtomicReference<AsyncJournalWriter> journalWriter); Journaled getStateMachine(); @Override URI getLocation(); @Override JournalContext createJournalContext(); @Override void close();  }