[FLINK-8132][kafka] Re-initialize transactional KafkaProducer on each checkpoint
Previously faulty scenario with producer pool of 2.
1. started transaction 1 with producerA, written record 42
2. checkpoint 1 triggered, pre committing txn1, started txn2 with producerB,
written record 43
3.
[FLINK-8132][kafka] Re-initialize transactional KafkaProducer on each checkpoint
Previously faulty scenario with producer pool of 2.
1. started transaction 1 with producerA, written record 42
2. checkpoint 1 triggered, pre committing txn1, started txn2 with producerB,
written record 43
3.
