Hello Kafka! I'm using Flink with a Kafka and seeing an edge case that I'm hoping to get some help to find the source code that would explain!
In the Flink setup, I have a transactional producer that's regularly committing transactions with a consistent transactional ID. But all these transactions have no data records written into them. Based on my understanding according to the transactions design doc. After transactional.id.expiration.ms has passed, because there have been transactions committed regularly, the transactional id should not expire, and the transactionalId mapping would contain the PID. But since there is no record written for that producer, it's possible that the PID is expired. We are seeing endTxnRequests failing with invalidMappingException. But looking from the source code, since the transactional id entry should still exist. I don't see why this could be the reason that an InvalidMappingException is thrown. My questions are: 1. Is it correct that if transactions are being regularly committed, transactional id should not expire even if there is no data written by the producer? 2. what failures can be triggered when a transactional id has NOT expired, but the associated producer id has expired? 3. Do the expiration of producer id trigger a cleanup task to remove the transactional id entry from the PID snapshot file? Thank you!
