Default Kafka producers pool size for FlinkKafkaProducer.Semantic.EXACTLY_ONCE

[min.tan]

Hi,

 

I keep getting exceptions "org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints."

 

I think that DEFAULT_KAFKA_PRODUCERS_POOL_SIZE is 5 and need to increase this size. What considerations should I take to increase this size?

 

I have a check point setting like this and run a parallelism of 16 and have a check point setting like this

 

public static void setup(StreamExecutionEnvironment env) {
    env.enableCheckpointing(2_000);
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1_000);
    env.getCheckpointConfig().setCheckpointTimeout(10_000);
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
    env.setStateBackend(new MemoryStateBackend(Integer.MAX_VALUE));
    //env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
}

 

Regards,

 

Min