@Test
public void parallelSessions() {

    // Suppressing logs due to stacktrace for test4. Comment out to see logs.
    actorSystem.eventStream().setLogLevel(Attributes.logLevelOff());
    final ActorRef underTest = createSubscriptionManager();
    final TestProbe probe1 = TestProbe.apply("test1", actorSystem);
    final TestProbe probe2 = TestProbe.apply("test2", actorSystem);
    final TestProbe probe3 = TestProbe.apply("test3", actorSystem);
    final TestProbe probe4 = TestProbe.apply("test4", actorSystem);
    // expect success with results
    underTest.tell(createSubscription(1), probe1.ref());
    // expect success with prefix
    final String prefix = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
    underTest.tell(createSubscription(2).setPrefix(prefix), probe2.ref());
    // expect cancellation
    underTest.tell(createSubscription(3), probe3.ref());
    // expect upstream failure
    underTest.tell(createSubscription(4), probe4.ref());
    final String sid1 = probe1.expectMsgClass(SubscriptionCreated.class).getSubscriptionId();
    final String sid2 = probe2.expectMsgClass(SubscriptionCreated.class).getSubscriptionId();
    final String sid3 = probe3.expectMsgClass(SubscriptionCreated.class).getSubscriptionId();
    final String sid4 = probe4.expectMsgClass(SubscriptionCreated.class).getSubscriptionId();
    final List<?> sources = List.of(
            Source.single("t:1"),
            Source.single("t:2"),
            Source.from(List.of("t:3", "t:4")),
            new AskTimeoutException("mock error")
    );
    underTest.tell(request(sid1), probe1.ref());
    underTest.tell(request(sid2), probe2.ref());
    underTest.tell(request(sid3), probe3.ref());
    underTest.tell(request(sid4), probe4.ref());
    // there should be no upstream request until downstream requests.
    for (int i = 0; i < 4; ++i) {
        final StreamThings streamThings = conciergeForwarderProbe.expectMsgClass(StreamThings.class);
        final ActorRef sender = conciergeForwarderProbe.sender();
        final Object source = sources.get(getTag(streamThings) - 1);
        if (source instanceof Source) {
            final SourceRef<?> sourceRef = ((Source<?, ?>) source).runWith(StreamRefs.sourceRef(), materializer);
            sender.tell(sourceRef, ActorRef.noSender());
        } else {
            sender.tell(source, ActorRef.noSender());
        }
    }
    probe1.expectMsg(hasNext(sid1, "t:1"));
    probe1.expectMsg(subscriptionComplete(sid1));
    probe2.expectMsg(hasNext(sid2, "t:2"));
    probe2.expectMsg(subscriptionComplete(sid2));
    probe3.expectMsg(hasNext(sid3, "t:3"));
    underTest.tell(CancelSubscription.of(sid3, DittoHeaders.empty()), probe3.ref());
    assertThat(probe4.expectMsgClass(SubscriptionFailed.class).getError())
            .isInstanceOf(GatewayInternalErrorException.class);
    CompletableFuture.allOf(
            CompletableFuture.runAsync(
                    () -> probe3.expectNoMessage(FiniteDuration.apply(250L, TimeUnit.MILLISECONDS))),
            CompletableFuture.runAsync(
                    () -> probe4.expectNoMessage(FiniteDuration.apply(250L, TimeUnit.MILLISECONDS)))
    ).join();
}