How to use JUnit to test asynchronous processes

Question

How do you test methods that fire asynchronous processes with JUnit?

I don't know how to make my test wait for the process to end (it is not exactly a unit test, it is more like an integration test as it involves several classes and not just one).

Answer 1

TL;DR; Unfortunately, there is no built-in solution yet (at time of writting, 2022), hence you are free to use and/or implement whatever fits your situation.

Example

An alternative is to use the CountDownLatch class.

public class DatabaseTest {

    /**
     * Data limit
     */
    private static final int DATA_LIMIT = 5;

    /**
     * Countdown latch
     */
    private CountDownLatch lock = new CountDownLatch(1);

    /**
     * Received data
     */
    private List<Data> receiveddata;

    @Test
    public void testDataRetrieval() throws Exception {
        Database db = new MockDatabaseImpl();
        db.getData(DATA_LIMIT, new DataCallback() {
            @Override
            public void onSuccess(List<Data> data) {
                receiveddata = data;
                lock.countDown();
            }
        });

        lock.await(2000, TimeUnit.MILLISECONDS);

        assertNotNull(receiveddata);
        assertEquals(DATA_LIMIT, receiveddata.size());
    }
}

NOTE you can't just used syncronized with a regular object as a lock, as fast callbacks can release the lock before the lock's wait method is called. See this blog post by Joe Walnes.

EDIT Removed syncronized blocks around CountDownLatch thanks to comments from @jtahlborn and @Ring

Answer 2

You can try using the Awaitility library. It makes it easy to test the systems you're talking about.

Answer 3

If you use a CompletableFuture (introduced in Java 8) or a SettableFuture (from Google Guava), you can make your test finish as soon as it's done, rather than waiting a pre-set amount of time. Your test would look something like this:

CompletableFuture<String> future = new CompletableFuture<>();
executorService.submit(new Runnable() {         
    @Override
    public void run() {
        future.complete("Hello World!");                
    }
});
assertEquals("Hello World!", future.get());

Note that there is a library which Provides CompletableFuture for pre Java-8, which even uses the same names (and provides all related Java-8 classes), like:

net.sourceforge.streamsupport:streamsupport-minifuture:1.7.4

This is useful for Android development, where even if we build with JDK-v11, we want to keep codes compatible with pre Android-7 devices.

Answer 4

IMHO it's bad practice to have unit tests create or wait on threads, etc. You'd like these tests to run in split seconds. That's why I'd like to propose a 2-step approach to testing async processes.

1. Test that your async process is submitted properly. You can mock the object that accepts your async requests and make sure that the submitted job has correct properties, etc.
2. Test that your async callbacks are doing the right things. Here you can mock out the originally submitted job and assume it's initialized properly and verify that your callbacks are correct.

Answer 5

Start the process off and wait for the result using a Future.

Answer 6

One method I've found pretty useful for testing asynchronous methods is injecting an Executor instance in the object-to-test's constructor. In production, the executor instance is configured to run asynchronously while in test it can be mocked to run synchronously.

So suppose I'm trying to test the asynchronous method Foo#doAsync(Callback c),

class Foo {
  private final Executor executor;
  public Foo(Executor executor) {
    this.executor = executor;
  }

  public void doAsync(Callback c) {
    executor.execute(new Runnable() {
      @Override public void run() {
        // Do stuff here
        c.onComplete(data);
      }
    });
  }
}

In production, I would construct Foo with an Executors.newSingleThreadExecutor() Executor instance while in test I would probably construct it with a synchronous executor that does the following --

class SynchronousExecutor implements Executor {
  @Override public void execute(Runnable r) {
    r.run();
  }
}

Now my JUnit test of the asynchronous method is pretty clean --

@Test public void testDoAsync() {
  Executor executor = new SynchronousExecutor();
  Foo objectToTest = new Foo(executor);

  Callback callback = mock(Callback.class);
  objectToTest.doAsync(callback);

  // Verify that Callback#onComplete was called using Mockito.
  verify(callback).onComplete(any(Data.class));

  // Assert that we got back the data that we expected.
  assertEquals(expectedData, callback.getData());
}

Answer 7

JUnit 5 has Assertions.assertTimeout(Duration, Executable)/assertTimeoutPreemptively()(please read Javadoc of each to understand the difference) and Mockito has verify(mock, timeout(millisecs).times(x)).

Assertions.assertTimeout(Duration.ofMillis(1000), () -> 
    myReactiveService.doSth().subscribe()
);

And:

Mockito.verify(myReactiveService, 
    timeout(1000).times(0)).doSth(); // cannot use never() here

Timeout may be nondeterministic/fragile in pipelines. So be careful.

Answer 8

How about calling SomeObject.wait and notifyAll as described here OR using Robotiums Solo.waitForCondition(...) method OR use a class i wrote to do this (see comments and test class for how to use)

Answer 9

I find an library socket.io to test asynchronous logic. It looks simple and brief way using LinkedBlockingQueue. Here is example:

    @Test(timeout = TIMEOUT)
public void message() throws URISyntaxException, InterruptedException {
    final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();

    socket = client();
    socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
        @Override
        public void call(Object... objects) {
            socket.send("foo", "bar");
        }
    }).on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
        @Override
        public void call(Object... args) {
            values.offer(args);
        }
    });
    socket.connect();

    assertThat((Object[])values.take(), is(new Object[] {"hello client"}));
    assertThat((Object[])values.take(), is(new Object[] {"foo", "bar"}));
    socket.disconnect();
}

Using LinkedBlockingQueue take API to block until to get result just like synchronous way. And set timeout to avoid assuming too much time to wait the result.

Answer 10

It's worth mentioning that there is very useful chapter Testing Concurrent Programs in Concurrency in Practice which describes some unit testing approaches and gives solutions for issues.

Answer 11

This is what I'm using nowadays if the test result is produced asynchronously.

public class TestUtil {

    public static <R> R await(Consumer<CompletableFuture<R>> completer) {
        return await(20, TimeUnit.SECONDS, completer);
    }

    public static <R> R await(int time, TimeUnit unit, Consumer<CompletableFuture<R>> completer) {
        CompletableFuture<R> f = new CompletableFuture<>();
        completer.accept(f);
        try {
            return f.get(time, unit);
        } catch (InterruptedException | TimeoutException e) {
            throw new RuntimeException("Future timed out", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Future failed", e.getCause());
        }
    }
}

Using static imports, the test reads kinda nice. (note, in this example I'm starting a thread to illustrate the idea)

    @Test
    public void testAsync() {
        String result = await(f -> {
            new Thread(() -> f.complete("My Result")).start();
        });
        assertEquals("My Result", result);
    }

If f.complete isn't called, the test will fail after a timeout. You can also use f.completeExceptionally to fail early.

Answer 12

There are many answers here but a simple one is to just create a completed CompletableFuture and use it:

CompletableFuture.completedFuture("donzo")

So in my test:

this.exactly(2).of(mockEventHubClientWrapper).sendASync(with(any(LinkedList.class)));
this.will(returnValue(new CompletableFuture<>().completedFuture("donzo")));

I am just making sure all of this stuff gets called anyway. This technique works if you are using this code:

CompletableFuture.allOf(calls.toArray(new CompletableFuture[0])).join();

It will zip right through it as all the CompletableFutures are finished!

Answer 13

Avoid testing with parallel threads whenever you can (which is most of the time). This will only make your tests flaky (sometimes pass, sometimes fail).

Only when you need to call some other library / system, you might have to wait on other threads, in that case always use the Awaitility library instead of Thread.sleep().

Never just call get() or join() in your tests, else your tests might run forever on your CI server in case the future never completes. Always assert isDone() first in your tests before calling get(). For CompletionStage, that is .toCompletableFuture().isDone().

When you test a non-blocking method like this:

public static CompletionStage<String> createGreeting(CompletableFuture<String> future) {
    return future.thenApply(result -> "Hello " + result);
}

then you should not just test the result by passing a completed Future in the test, you should also make sure that your method doSomething() does not block by calling join() or get(). This is important in particular if you use a non-blocking framework.

To do that, test with a non-completed future that you set to completed manually:

@Test
public void testDoSomething() throws Exception {
    CompletableFuture<String> innerFuture = new CompletableFuture<>();
    CompletableFuture<String> futureResult = createGreeting(innerFuture).toCompletableFuture();
    assertFalse(futureResult.isDone());

    // this triggers the future to complete
    innerFuture.complete("world");
    assertTrue(futureResult.isDone());

    // futher asserts about fooResult here
    assertEquals(futureResult.get(), "Hello world");
}

That way, if you add future.join() to doSomething(), the test will fail.

If your Service uses an ExecutorService such as in thenApplyAsync(..., executorService), then in your tests inject a single-threaded ExecutorService, such as the one from guava:

ExecutorService executorService = Executors.newSingleThreadExecutor();

If your code uses the forkJoinPool such as thenApplyAsync(...), rewrite the code to use an ExecutorService (there are many good reasons), or use Awaitility.

To shorten the example, I made BarService a method argument implemented as a Java8 lambda in the test, typically it would be an injected reference that you would mock.

Answer 14

For all Spring users out there, this is how I usually do my integration tests nowadays, where async behaviour is involved:

Fire an application event in production code, when an async task (such as an I/O call) has finished. Most of the time this event is necessary anyway to handle the response of the async operation in production.

With this event in place, you can then use the following strategy in your test case:

1. Execute the system under test
2. Listen for the event and make sure that the event has fired
3. Do your assertions

To break this down, you'll first need some kind of domain event to fire. I'm using a UUID here to identify the task that has completed, but you're of course free to use something else as long as it's unique.

(Note, that the following code snippets also use Lombok annotations to get rid of boiler plate code)

@RequiredArgsConstructor
class TaskCompletedEvent() {
  private final UUID taskId;
  // add more fields containing the result of the task if required
}

The production code itself then typically looks like this:

@Component
@RequiredArgsConstructor
class Production {

  private final ApplicationEventPublisher eventPublisher;

  void doSomeTask(UUID taskId) {
    // do something like calling a REST endpoint asynchronously
    eventPublisher.publishEvent(new TaskCompletedEvent(taskId));
  }

}

I can then use a Spring @EventListener to catch the published event in test code. The event listener is a little bit more involved, because it has to handle two cases in a thread safe manner:

1. Production code is faster than the test case and the event has already fired before the test case checks for the event, or
2. Test case is faster than production code and the test case has to wait for the event.

A CountDownLatch is used for the second case as mentioned in other answers here. Also note, that the @Order annotation on the event handler method makes sure, that this event handler method gets called after any other event listeners used in production.

@Component
class TaskCompletionEventListener {

  private Map<UUID, CountDownLatch> waitLatches = new ConcurrentHashMap<>();
  private List<UUID> eventsReceived = new ArrayList<>();

  void waitForCompletion(UUID taskId) {
    synchronized (this) {
      if (eventAlreadyReceived(taskId)) {
        return;
      }
      checkNobodyIsWaiting(taskId);
      createLatch(taskId);
    }
    waitForEvent(taskId);
  }

  private void checkNobodyIsWaiting(UUID taskId) {
    if (waitLatches.containsKey(taskId)) {
      throw new IllegalArgumentException("Only one waiting test per task ID supported, but another test is already waiting for " + taskId + " to complete.");
    }
  }

  private boolean eventAlreadyReceived(UUID taskId) {
    return eventsReceived.remove(taskId);
  }

  private void createLatch(UUID taskId) {
    waitLatches.put(taskId, new CountDownLatch(1));
  }

  @SneakyThrows
  private void waitForEvent(UUID taskId) {
    var latch = waitLatches.get(taskId);
    latch.await();
  }

  @EventListener
  @Order
  void eventReceived(TaskCompletedEvent event) {
    var taskId = event.getTaskId();
    synchronized (this) {
      if (isSomebodyWaiting(taskId)) {
        notifyWaitingTest(taskId);
      } else {
        eventsReceived.add(taskId);
      }
    }
  }

  private boolean isSomebodyWaiting(UUID taskId) {
    return waitLatches.containsKey(taskId);
  }

  private void notifyWaitingTest(UUID taskId) {
    var latch = waitLatches.remove(taskId);
    latch.countDown();
  }

}

Last step is to execute the system under test in a test case. I'm using a SpringBoot test with JUnit 5 here, but this should work the same for all tests using a Spring context.

@SpringBootTest
class ProductionIntegrationTest {

  @Autowired
  private Production sut;

  @Autowired
  private TaskCompletionEventListener listener;

  @Test
  void thatTaskCompletesSuccessfully() {
    var taskId = UUID.randomUUID();
    sut.doSomeTask(taskId);
    listener.waitForCompletion(taskId);
    // do some assertions like looking into the DB if value was stored successfully
  }

}

Note, that in contrast to other answers here, this solution will also work if you execute your tests in parallel and multiple threads exercise the async code at the same time.

Answer 15

I prefer use wait and notify. It is simple and clear.

@Test
public void test() throws Throwable {
    final boolean[] asyncExecuted = {false};
    final Throwable[] asyncThrowable= {null};

    // do anything async
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                // Put your test here.
                fail(); 
            }
            // lets inform the test thread that there is an error.
            catch (Throwable throwable){
                asyncThrowable[0] = throwable;
            }
            // ensure to release asyncExecuted in case of error.
            finally {
                synchronized (asyncExecuted){
                    asyncExecuted[0] = true;
                    asyncExecuted.notify();
                }
            }
        }
    }).start();

    // Waiting for the test is complete
    synchronized (asyncExecuted){
        while(!asyncExecuted[0]){
            asyncExecuted.wait();
        }
    }

    // get any async error, including exceptions and assertationErrors
    if(asyncThrowable[0] != null){
        throw asyncThrowable[0];
    }
}

Basically, we need to create a final Array reference, to be used inside of anonymous inner class. I would rather create a boolean[], because I can put a value to control if we need to wait(). When everything is done, we just release the asyncExecuted.

Answer 16

If you want to test the logic just don´t test it asynchronously.

For example to test this code which works on results of an asynchronous method.

public class Example {
    private Dependency dependency;

    public Example(Dependency dependency) {
        this.dependency = dependency;            
    }

    public CompletableFuture<String> someAsyncMethod(){
        return dependency.asyncMethod()
                .handle((r,ex) -> {
                    if(ex != null) {
                        return "got exception";
                    } else {
                        return r.toString();
                    }
                });
    }
}

public class Dependency {
    public CompletableFuture<Integer> asyncMethod() {
        // do some async stuff       
    }
}

In the test mock the dependency with synchronous implementation. The unit test is completely synchronous and runs in 150ms.

public class DependencyTest {
    private Example sut;
    private Dependency dependency;

    public void setup() {
        dependency = Mockito.mock(Dependency.class);;
        sut = new Example(dependency);
    }

    @Test public void success() throws InterruptedException, ExecutionException {
        when(dependency.asyncMethod()).thenReturn(CompletableFuture.completedFuture(5));

        // When
        CompletableFuture<String> result = sut.someAsyncMethod();

        // Then
        assertThat(result.isCompletedExceptionally(), is(equalTo(false)));
        String value = result.get();
        assertThat(value, is(equalTo("5")));
    }

    @Test public void failed() throws InterruptedException, ExecutionException {
        // Given
        CompletableFuture<Integer> c = new CompletableFuture<Integer>();
        c.completeExceptionally(new RuntimeException("failed"));
        when(dependency.asyncMethod()).thenReturn(c);

        // When
        CompletableFuture<String> result = sut.someAsyncMethod();

        // Then
        assertThat(result.isCompletedExceptionally(), is(equalTo(false)));
        String value = result.get();
        assertThat(value, is(equalTo("got exception")));
    }
}

You don´t test the async behaviour but you can test if the logic is correct.

Answer 17

Let's say you have this code:

public void method() {
        CompletableFuture.runAsync(() -> {
            //logic
            //logic
            //logic
            //logic
        });
    }

Try to refactor it to something like this:

public void refactoredMethod() {
    CompletableFuture.runAsync(this::subMethod);
}

private void subMethod() {
    //logic
    //logic
    //logic
    //logic
}

After that, test the subMethod this way:

org.powermock.reflect.Whitebox.invokeMethod(classInstance, "subMethod"); 

This isn't a perfect solution, but it tests all the logic inside your async execution.

Answer 18

The following solution is explained in the code, there are two different approaches depending on your needs.

package es.victorherraiz.learningjava;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import static java.util.concurrent.CompletableFuture.delayedExecutor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.timeout;

@ExtendWith(AsyncTest.AsyncExtension.class)
public class AsyncTest {


    /**
     * This first test uses a custom extension, the code is down there.
     * It is useful when there are plenty of async tests
     * @param cb The callback
     */
    @Test
    void test1(AsyncCallback cb) {
        cb.setTimeout(5000);
        delayedExecutor(2000, MILLISECONDS)
            .execute(cb::success);
    }

    /**
     * This test uses mockito, it is slower to detect signal.
     * Use this when there are only a few async tests.
     */
    @Test
    void test2() {
        var mock = mock(Runnable.class);
        delayedExecutor(2000, MILLISECONDS)
            .execute(mock);
        then(mock).should(timeout(5000)).run();
    }

    public static final class AsyncCallback {
        private final CountDownLatch latch = new CountDownLatch(1);
        private final AtomicLong timeout = new AtomicLong(5000);
        private Exception exception;

        private void setTimeout(long timeout) {
            this.timeout.set(timeout);
        }

        private long getTimeout() {
            return this.timeout.get();
        };


        boolean await() throws InterruptedException {
            return latch.await(timeout.get(), MILLISECONDS);
        }

        Exception getFailure() {
            return exception;
        }

        public void done(Exception exception) {
            this.exception = exception;
            latch.countDown();
        }

        public void success() {
            done(null);
        }

        public void failure(Exception exception) {
            done(exception);
        }
    }

    // Extract this class to any common package
    public static class AsyncExtension implements ParameterResolver, AfterTestExecutionCallback {
        private static final ExtensionContext.Namespace NAMESPACE =
            ExtensionContext.Namespace.create(AsyncExtension.class);

        @Override
        public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
            return parameterContext.getParameter().getType().equals(AsyncCallback.class);
        }

        @Override
        public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
            return extensionContext.getStore(NAMESPACE).getOrComputeIfAbsent(AsyncCallback.class);
        }

        @Override
        public void afterTestExecution(ExtensionContext extensionContext) throws Exception {
            var callback = (AsyncCallback) extensionContext.getStore(NAMESPACE).get(AsyncCallback.class);
            if (callback != null) {
                if (!callback.await()) {
                    throw new TimeoutException("Timeout: there was not signal for " + callback.getTimeout() + "ms");
                }
                var failure = callback.getFailure();
                if (failure != null) {
                    throw failure;
                }
            }
        }
    }

}