Re: New kafka producer on each checkpoint

2020-04-13 Thread Becket Qin
A slightly more common case that may cause the producer to be not reusable is when there is no data for long time, the producer won't send any request to the broker and the tansactional.id may also expire on the broker side. On Tue, Apr 14, 2020 at 8:44 AM Becket Qin wrote: > Hi Maxim, > > That

Re: New kafka producer on each checkpoint

2020-04-13 Thread Becket Qin
Hi Maxim, That is a good question. I don't see an obvious reason that we cannot reuse the producers. That said, there might be some corner cases where the KafkaProducer is not reusable. For example, if the checkpoint interval is long, the producer.id assigned by the TransactionCoordinator may have

Re: New kafka producer on each checkpoint

2020-04-13 Thread Maxim Parkachov
Hi Yun, thanks for the answer. I did now increased checkpoint interval, but still I don't understand reason for creating producer and re-connecting to to kafka broker each time. According to documentation: Note: Semantic.EXACTLY_ONCE mode uses a fixed size pool of KafkaProducers per each FlinkKaf

Re: New kafka producer on each checkpoint

2020-04-07 Thread Yun Tang
Hi Maxim If you use the EXACTLY_ONCE semantic (instead of AT_LEAST_ONCE or NONE) for flink kafka producer. It will create new producer when every new checkpoint comes [1]. This is by design and from my point of view, the checkpoint interval of 10 seconds might be a bit too often. In general I t