[FLINK-22452] Support specifying custom transactional.id prefix in FlinkKafkaProducer

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.12.2
• Fix Version/s: 1.14.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Wenhao Ji
• Reporter: Wenhao Ji
• Created: 25/Apr/21 09:14
• Updated: 23/Sep/21 18:04
• Resolved: 13/Aug/21 06:54

Description

Currently, the "transactional.id"s of the Kafka producers in FlinkKafkaProducer are generated based on the task name. This mechanism has some limitations:

• It will exceed Kafka's limitation if the task name is too long. (resolved in FLINK-17691)
• They will very likely clash each other if the job topologies are similar. (discussed in FLINK-11654)
• Only certain "transactional.id" may be authorized by Prefixed ACLs on the target Kafka cluster.

Besides, the spring community has introduced the setTransactionIdPrefix method to their Kafka client.

Therefore, I think it will be necessary to have this feature in the Flink Kafka connector. 

 

As discussed in FLINK-11654, the possible solution will be,

• either introduce an additional method called setTransactionalIdPrefix(String) in the FlinkKafkaProducer,
• or use the existing "transactional.id" properties as the prefix.

 And the behavior of the "transactional.id" generation will be

• keep the behavior as it was if absent,
• use the one if present as the prefix for the TransactionalIdsGenerator.

Issue Links

• is related to: FLINK-17691 FlinkKafkaProducer transactional.id too long when using Semantic.EXACTLY_ONCE
• is related to: FLINK-11654 Multiple transactional KafkaProducers writing to same cluster have clashing transaction IDs
• relates to: FLINK-11654 Multiple transactional KafkaProducers writing to same cluster have clashing transaction IDs
• links to: GitHub Pull Request #16494
• links to: GitHub Pull Request #16746