### Question:
TaskHistoryWriter extends AbstractIdleService implements Runnable { public void saveHistoryItem(final TaskStatus status) throws InterruptedException { saveHistoryItem(status, System.currentTimeMillis()); } TaskHistoryWriter(final String hostname, final ZooKeeperClient client,
                           final Path backingFile); void saveHistoryItem(final TaskStatus status); void saveHistoryItem(final TaskStatus status, long timestamp); boolean isEmpty(); @Override void run(); @VisibleForTesting
  static final int MAX_NUMBER_STATUS_EVENTS_TO_RETAIN; }### Answer:
@Test public void testZooKeeperErrorDoesntLoseItemsReally() throws Exception { final ZooKeeperClient mockClient = mock(ZooKeeperClient.class, delegatesTo(client)); final String path = Paths.historyJobHostEventsTimestamp(JOB_ID, HOSTNAME, TIMESTAMP); final AtomicBoolean throwExceptionOnCreateAndSet = new AtomicBoolean(true); final KeeperException exc = new ConnectionLossException(); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { if (throwExceptionOnCreateAndSet.get()) { throw exc; } else { client.createAndSetData( (String) invocation.getArguments()[0], (byte[]) invocation.getArguments()[1]); return null; } } }).when(mockClient).createAndSetData(path, TASK_STATUS.toJsonBytes()); makeWriter(mockClient); writer.saveHistoryItem(TASK_STATUS, TIMESTAMP); verify(mockClient, timeout(10000).atLeast(2)).createAndSetData(path, TASK_STATUS.toJsonBytes()); throwExceptionOnCreateAndSet.set(false); awaitHistoryItems(); }
@Test public void testSimpleWorkage() throws Exception { writer.saveHistoryItem(TASK_STATUS, TIMESTAMP); final TaskStatusEvent historyItem = Iterables.getOnlyElement(awaitHistoryItems()); assertEquals(JOB_ID, historyItem.getStatus().getJob().getId()); }
@Test public void testWriteWithZooKeeperDown() throws Exception { zk.stop(); writer.saveHistoryItem(TASK_STATUS, TIMESTAMP); zk.start(); final TaskStatusEvent historyItem = Iterables.getOnlyElement(awaitHistoryItems()); assertEquals(JOB_ID, historyItem.getStatus().getJob().getId()); }
@Test public void testWriteWithZooKeeperDownAndInterveningCrash() throws Exception { zk.stop(); writer.saveHistoryItem(TASK_STATUS, TIMESTAMP); writer.stopAsync().awaitTerminated(); makeWriter(client); zk.start(); final TaskStatusEvent historyItem = Iterables.getOnlyElement(awaitHistoryItems()); assertEquals(JOB_ID, historyItem.getStatus().getJob().getId()); }