@Test public void gainPrimacyDuringCatchup() throws Exception { CountingDummyFileSystemMaster countingMaster = new CountingDummyFileSystemMaster(); mFollowerJournalSystem.createJournal(countingMaster); final int entryCount = 100000; mFollowerJournalSystem.suspend(); final long catchupIndex = entryCount - 5; Map<String, Long> backupSequences = new HashMap<>(); backupSequences.put("FileSystemMaster", catchupIndex); CatchupFuture catchupFuture = mFollowerJournalSystem.catchup(backupSequences); ForkJoinPool.commonPool().submit(() -> { try (JournalContext journalContext = mLeaderJournalSystem.createJournal(new NoopMaster()).createJournalContext()) { for (int i = 0; i < entryCount; i++) { journalContext .append( alluxio.proto.journal.Journal.JournalEntry.newBuilder() .setInodeLastModificationTime( File.InodeLastModificationTimeEntry.newBuilder().setId(i).build()) .build()); } } catch (Exception e) { Assert.fail(String.format("Failed while writing entries: %s", e.toString())); } }); CommonUtils.waitFor("Advancing to start.", () -> countingMaster.getApplyCount() > 0, mWaitOptions); mLeaderJournalSystem.notifyLeadershipStateChanged(false); mFollowerJournalSystem.notifyLeadershipStateChanged(true); mFollowerJournalSystem.gainPrimacy(); CommonUtils.waitFor( "full state acquired after resume", () -> mFollowerJournalSystem.getCurrentSequenceNumbers() .values().stream().distinct().collect(Collectors.toList()).get(0) == entryCount - 1, mWaitOptions); mThrown.expect(IllegalStateException.class); mFollowerJournalSystem.resume(); }
@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();  }