Bulkhead policy may drop requests when maxWaitTime is specified about failsafe HOT 8 CLOSED

  public void test() throws Throwable {
    Bulkhead<Object> bulkhead = Bulkhead.of(1);
    List<CompletableFuture<Object>> stages = new ArrayList<>();

    Waiter waiter = new Waiter();
    for (int i = 0; i < 3; i++) {
      stages.add(Failsafe.with(bulkhead).getStageAsync(() -> CompletableFuture.supplyAsync(() -> {
        Testing.sleep(10);
        waiter.resume();
        return null;
      })));
      Testing.sleep(200); // Needed to avoid BulkheadFullException
    }

    waiter.await(1, TimeUnit.MINUTES, 3);
    for (CompletableFuture<Object> stage : stages)
      assertTrue(stage.isDone());
  }

package com.example.failsafe;

import dev.failsafe.*;
import java.time.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.junit.*;
import static org.junit.Assert.*;


/**
 * See https://github.com/failsafe-lib/failsafe/issues/365
 */
public class Issue365 {

    static final int NUM_TASKS = 3;
    static final long MAX_WAIT_SECS = 10;
    static final long TASK_SLEEP_SECS = 1;
    static final int CONCURRENCY = 1;

    Bulkhead<Integer> BULKHEAD = Bulkhead.<Integer>builder(CONCURRENCY)
        .withMaxWaitTime(Duration.ofSeconds(MAX_WAIT_SECS))
        .build();

    FailsafeExecutor<Integer> EXEC = Failsafe.with(BULKHEAD);


    @Test public void threeTasks() {
        AtomicInteger count = new AtomicInteger(NUM_TASKS);
        try {
            CompletableFuture<?>[] tasks = new CompletableFuture<?>[NUM_TASKS];
            for (int id = 0; id < NUM_TASKS; ++id) {
                int ID = id;
                tasks[ID] = EXEC.getStageAsync(() ->
                    CompletableFuture.supplyAsync(() -> sleepThenReturn(count, ID)));
            }
            CompletableFuture.allOf(tasks).get();
        } catch (InterruptedException ex) {
            System.out.println("interrupted get");
        } catch (ExecutionException ex) {
            if (ex.getCause() instanceof BulkheadFullException) {
                System.out.printf("Bulkhead full: %s%n", ex.getCause());
            }
        }

        assertEquals(0, count.get());
    }

    private Integer sleepThenReturn(AtomicInteger count, int id) {
        try {
            TimeUnit.SECONDS.sleep(TASK_SLEEP_SECS);
            int remaining = count.decrementAndGet();
            System.out.printf("task %d done sleeping, %d remaining%n", id, remaining);
            return id;
        } catch (InterruptedException ex) {
            System.out.println("interrupted sleep");
            return -1;
        }
    }
}

public synchronized void releasePermit() {
    if (permits < maxPermits) {
      permits += 1;
      CompletableFuture<Void> future = futures.pollFirst();
      if (future != null)
        future.complete(null);
    }
  }

public synchronized void releasePermit() {
    if (permits < maxPermits) {
      permits += 1;
      CompletableFuture<Void> future = futures.pollFirst();
      if (future != null){
        permits -= 1;
        future.complete(null);
      }
    }
  }