@Test
public void testLargePayloadCanFailUpdateWithRetry() {

    Task task = testTask();
    Worker worker = mock(Worker.class);
    when(worker.getPollingInterval()).thenReturn(3000);
    when(worker.getTaskDefName()).thenReturn(TEST_TASK_DEF_NAME);
    when(worker.execute(any())).thenReturn(new TaskResult(task));
    TaskClient taskClient = Mockito.mock(TaskClient.class);
    when(taskClient.pollTask(any(), any(), any())).thenReturn(task);
    when(taskClient.ack(any(), any())).thenReturn(true);
    doAnswer(invocation -> {
        Object[] args = invocation.getArguments();
        TaskResult result = (TaskResult) args[0];
        assertNull(result.getReasonForIncompletion());
        result.setReasonForIncompletion("some_reason");
        throw new ConductorClientException();
    }).when(taskClient).evaluateAndUploadLargePayload(any(TaskResult.class), any());
    TaskPollExecutor taskPollExecutor = new TaskPollExecutor(null, taskClient, 1, 3, new HashMap<>(), "test-worker-");
    CountDownLatch latch = new CountDownLatch(1);
    doAnswer(invocation -> {
            latch.countDown();
            return null;
        }
    ).when(worker).onErrorUpdate(any());
    Executors.newSingleThreadScheduledExecutor()
        .scheduleAtFixedRate(() -> taskPollExecutor.pollAndExecute(worker), 0, 1, TimeUnit.SECONDS);
    Uninterruptibles.awaitUninterruptibly(latch);
    // When evaluateAndUploadLargePayload fails indefinitely, task update shouldn't be called.
    verify(taskClient, times(0)).updateTask(any());
}