@Test public void canRetryBatchException() { final AtomicInteger attempt = new AtomicInteger(0); final List<String> errors = new LinkedList<>(); new WriteWithErrors(msg -> { if (attempt.incrementAndGet() == 1) { throw BatchException.of(new RuntimeException(), 0); } throw new RuntimeException(); }, msg -> { errors.add(msg.getMessageId()); return CompletableFuture.completedFuture(null); }, 2).apply(PubsubMessage.newBuilder().setMessageId("1").build()).join(); assertEquals(3, attempt.get()); assertEquals(ImmutableList.of("1"), errors); }
@Override public CompletableFuture<Void> apply(PubsubMessage message) { return apply(message, 1); }
WriteWithErrors implements Function<PubsubMessage, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(PubsubMessage message) { return apply(message, 1); } }
WriteWithErrors implements Function<PubsubMessage, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(PubsubMessage message) { return apply(message, 1); } WriteWithErrors(Function<PubsubMessage, CompletableFuture<Void>> write,
      Function<PubsubMessage, CompletableFuture<Void>> errors, int maxAttempts); }
WriteWithErrors implements Function<PubsubMessage, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(PubsubMessage message) { return apply(message, 1); } WriteWithErrors(Function<PubsubMessage, CompletableFuture<Void>> write,
      Function<PubsubMessage, CompletableFuture<Void>> errors, int maxAttempts); @Override CompletableFuture<Void> apply(PubsubMessage message); }
WriteWithErrors implements Function<PubsubMessage, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(PubsubMessage message) { return apply(message, 1); } WriteWithErrors(Function<PubsubMessage, CompletableFuture<Void>> write,
      Function<PubsubMessage, CompletableFuture<Void>> errors, int maxAttempts); @Override CompletableFuture<Void> apply(PubsubMessage message);  }