@Test
public void testWithDeliveryDelayAndMode() throws JMSException {

    MapBasedConfig config = new MapBasedConfig.Builder()
            .put("destination", "queue-one")
            .put("delivery-delay", 1500L)
            .put("delivery-mode", "non_persistent")
            .put("channel-name", "jms")
            .build();
    JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor);
    MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one"));
    subscriber = sink.getSink().build();
    subscriber.onSubscribe(new Subscriptions.EmptySubscription());
    AtomicBoolean acked = new AtomicBoolean();
    subscriber.onNext(Message.of("hello",
            () -> CompletableFuture.runAsync(() -> acked.set(true))));
    await()
            .atLeast(1, TimeUnit.SECONDS)
            .until(() -> client.messages.size() >= 1);
    assertThat(acked).isTrue();
    assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello");
    assertThat(client.messages.get(0).getJMSMessageID()).isNotNull();
    assertThat(client.messages.get(0).getJMSTimestamp()).isPositive();
}