@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(); }
public void saveHistoryItem(final TaskStatus status) throws InterruptedException { saveHistoryItem(status, System.currentTimeMillis()); }
TaskHistoryWriter extends AbstractIdleService implements Runnable { public void saveHistoryItem(final TaskStatus status) throws InterruptedException { saveHistoryItem(status, System.currentTimeMillis()); } }
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); }
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(); }
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; }