public void logAndProcessFailure(
      String computationId,
      ExecutableWork executableWork,
      Throwable t,
      Consumer<Work> onInvalidWork) {
    if (shouldRetryLocally(computationId, executableWork.work(), t)) {
      // Try again after some delay and at the end of the queue to avoid a tight loop.
      executeWithDelay(retryLocallyDelayMs, executableWork);
    } else {
      // Consider the item invalid. It will eventually be retried by Windmill if it still needs to
      // be processed.
      onInvalidWork.accept(executableWork.work());
    }
  }
@Test
  public void logAndProcessFailure_retriesOnUncaughtUnhandledException_streamingEngine()
      throws InterruptedException {
    CountDownLatch runWork = new CountDownLatch(1);
    ExecutableWork work = createWork(ignored -> runWork.countDown());
    WorkFailureProcessor workFailureProcessor =
        createWorkFailureProcessor(streamingEngineFailureReporter());
    Set<Work> invalidWork = new HashSet<>();
    workFailureProcessor.logAndProcessFailure(
        DEFAULT_COMPUTATION_ID, work, new RuntimeException(), invalidWork::add);

    runWork.await();
    assertThat(invalidWork).isEmpty();
  }