@Test public void restoreInvalidJournalEntryCheckpoint() throws IOException { Journaled journaled = new TestJournaled(0); ByteArrayOutputStream baos = new ByteArrayOutputStream(); CheckpointOutputStream cos = new CheckpointOutputStream(baos, CheckpointType.COMPOUND); cos.flush(); mThrown.expect(IllegalStateException.class); mThrown.expectMessage("Unrecognized checkpoint type"); JournalUtils.restoreJournalEntryCheckpoint( new CheckpointInputStream(new ByteArrayInputStream(baos.toByteArray())), journaled); }
public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled) throws IOException { Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY, "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(), input.getType()); journaled.resetState(); LOG.info("Reading journal entries"); JournalEntryStreamReader reader = new JournalEntryStreamReader(input); JournalEntry entry; while ((entry = reader.readEntry()) != null) { try { journaled.processJournalEntry(entry); } catch (Throwable t) { handleJournalReplayFailure(LOG, t, "Failed to process journal entry %s from a journal checkpoint", entry); } } }
JournalUtils { public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled) throws IOException { Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY, "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(), input.getType()); journaled.resetState(); LOG.info("Reading journal entries"); JournalEntryStreamReader reader = new JournalEntryStreamReader(input); JournalEntry entry; while ((entry = reader.readEntry()) != null) { try { journaled.processJournalEntry(entry); } catch (Throwable t) { handleJournalReplayFailure(LOG, t, "Failed to process journal entry %s from a journal checkpoint", entry); } } } }
JournalUtils { public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled) throws IOException { Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY, "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(), input.getType()); journaled.resetState(); LOG.info("Reading journal entries"); JournalEntryStreamReader reader = new JournalEntryStreamReader(input); JournalEntry entry; while ((entry = reader.readEntry()) != null) { try { journaled.processJournalEntry(entry); } catch (Throwable t) { handleJournalReplayFailure(LOG, t, "Failed to process journal entry %s from a journal checkpoint", entry); } } } private  JournalUtils(); }
JournalUtils { public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled) throws IOException { Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY, "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(), input.getType()); journaled.resetState(); LOG.info("Reading journal entries"); JournalEntryStreamReader reader = new JournalEntryStreamReader(input); JournalEntry entry; while ((entry = reader.readEntry()) != null) { try { journaled.processJournalEntry(entry); } catch (Throwable t) { handleJournalReplayFailure(LOG, t, "Failed to process journal entry %s from a journal checkpoint", entry); } } } private  JournalUtils(); static URI getJournalLocation(); static void writeJournalEntryCheckpoint(OutputStream output, JournalEntryIterable iterable); static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled); static void writeToCheckpoint(OutputStream output, List<? extends Checkpointed> components); static void restoreFromCheckpoint(CheckpointInputStream input,
      List<? extends Checkpointed> components); static void handleJournalReplayFailure(Logger logger, Throwable t,
      String format, Object... args); static void sinkAppend(Supplier<Set<JournalSink>> journalSinks, JournalEntry entry); static void sinkFlush(Supplier<Set<JournalSink>> journalSinks); }
JournalUtils { public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled) throws IOException { Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY, "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(), input.getType()); journaled.resetState(); LOG.info("Reading journal entries"); JournalEntryStreamReader reader = new JournalEntryStreamReader(input); JournalEntry entry; while ((entry = reader.readEntry()) != null) { try { journaled.processJournalEntry(entry); } catch (Throwable t) { handleJournalReplayFailure(LOG, t, "Failed to process journal entry %s from a journal checkpoint", entry); } } } private  JournalUtils(); static URI getJournalLocation(); static void writeJournalEntryCheckpoint(OutputStream output, JournalEntryIterable iterable); static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled); static void writeToCheckpoint(OutputStream output, List<? extends Checkpointed> components); static void restoreFromCheckpoint(CheckpointInputStream input,
      List<? extends Checkpointed> components); static void handleJournalReplayFailure(Logger logger, Throwable t,
      String format, Object... args); static void sinkAppend(Supplier<Set<JournalSink>> journalSinks, JournalEntry entry); static void sinkFlush(Supplier<Set<JournalSink>> journalSinks);  }