@Test(timeOut = 20000)
    public void testAbortDelayedMessage() throws Exception {
        final String topic = "persistent://my-property/my-ns/testAbortDelayedMessage";

        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic)
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();

        Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(false).create();

        admin.topicPolicies().setDelayedDeliveryPolicy(topic,
                DelayedDeliveryPolicies.builder()
                        .active(true)
                        .tickTime(100L)
                        .maxDeliveryDelayInMillis(5000)
                        .build());

        MessageId msgId = producer.newMessage()
                .value("Test delayed message".getBytes())
                .deliverAfter(5, TimeUnit.SECONDS)
                .send();
        log.warn("Message sent with ID: " + msgId);
        producer.flush();

        consumer.acknowledge(msgId);

        while (true) {
            Message<byte[]> msg = consumer.receive(100, TimeUnit.MILLISECONDS);
            if (msg != null) {
                log.warn("Received message: " + new String(msg.getData()));
                consumer.acknowledge(msg.getMessageId());
            }
            log.warn("Waiting for messages...");
        }
    }