[BUG] Flink pulsar sink delayed message incompatible with DeliveryGuarantee.EXACTLY_ONCE

[predator4ann]

Describe the bug
When use PulsarSink with delivery guarantee DeliveryGuarantee.EXACTLY_ONCE and delaySendingMessage, the deliverAtTime would be invalid and consumer can poll message before deliverAtTime.

To Reproduce
Steps to reproduce the behavior:

1. SinkDemo

PulsarSink.builder()
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setServiceUrl(serviceUrl)
            .setAdminUrl(adminUrl)
            .setSerializationSchema(serializationSchema)
            .setTopics(topic)
            .setProperties(props)
            .delaySendingMessage(MessageDelayer.fixed(Duration.ofMinutes(1)))
            .build();

2. SourceDemo

PulsarSource.builder()
            .setServiceUrl(serviceUrl)
            .setAdminUrl(adminUrl)
            .setStartCursor(StartCursor.earliest())
            .setDeserializationSchema(deserializationSchema)
            .setTopics(topic)
            .setSubscriptionName("my-subscription")
            .setSubscriptionType(SubscriptionType.Shared)
            .build();

Expected behavior
SourceDemo should receive message after at least 1 minute, actually pulsar source receive message before deliverAtTime in message metadata.

Screenshots
N/A

Additional context
Java client api : org.apache.flink:flink-connector-pulsar:1.15.1
Deploy mode: standalone on docker(2.10.1) & cluster on k8s(2.9.x)

[syhily]

Can you provided your environment?

• Flink version
• Pulsar version
• Connector version
• Connector artifact

[syhily]

It should be fixed by: apache/pulsar#17534

[syhily]

The PR is ready: apache/pulsar#17548

[syhily]

We will get this fixed after bump the client to 2.11.0

[predator4ann]

The PR is merged: apache/pulsar#17548