Hi, we are using Flink 1.8 with Pulsar as Source in our Production with the below configurations 1. disabled checkpointing 2. Using Subscription type Failover 3. acknowledgementBatchSize of 1 till now I used to think when the consumed message reaches the sink, Flink sends ACK to the pulsar. But looks like there is a default time interval after which it sends https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#message-acknowledgement Question: 1. Can someone please tell me what is the default value of PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL ? 2. Where exactly to set this? using java

PulsarSourceBuilder<Object> pulsarSourceBuilder = PulsarSourceBuilder
                .builder(new BeamPulsarSchema(Object.class))
                .pulsarAllClientConf(clientConfigurationData)
                .pulsarAllConsumerConf(consumerConfigurationData)
                .acknowledgementBatchSize(1)

Are you sure you are on Flink 1.8?

I'm asking because in later Flink versions a new Pulsar source has been added

okay, in new versions how does ack happens for this condition?

I'm not sure that this still exists in newer versions

Specifically you may want to look at https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/pulsar/#pulsar-source-enableautoacknowledgemessage.

got it I should use 2 things then? 1. pulsar.source.enableAutoAcknowledgeMessage = true 2. pulsar.source.autoCommitCursorInterval = desired time range after which we want Flink to send ACK to pulsar

yes