Status

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Kafka has introduced the Prefixed ACLs feature, by which producers may only be granted permissions to use "transactional.id"s with certain prefixes on a shared multiple-tenant Kafka cluster. Currently, the FlinkKafkaProducer generates "transactional.id" based on the task name and the operator's uid, which makes it hard and not straightforward to determine the "transactional.id" prefix ACL that should be onboard-ed on a Kafka cluster.

The purpose of this improvement is to let the "transactional.id" prefix that is generated and used by FlinkKafkaProducer be deterministic and customizable.

Public Interfaces

A new method `setTransactionalPrefix(String)` is added to the FlinkKafkaProducer class.

public class FlinkKafkaProducer {

    /**
      * Specify the transactional.id prefix to be used by the producers when communicating with Kafka.
      *
      * @param transactionalIdPrefix the transactional.id prefix
      */
    public void setTransactionalIdPrefix(String transactionalIdPrefix);
}

Proposed Changes

FlinkKafkaProducer

The FlinkKafkaProducer uses the TransactionalIdsGenerator class to generate the transactional id during its state initialization. By introducing the new method `setTransactionalPrefix`, the prefix, the first argument of the TransactionalIdsGenerator's constructor, will be,

• taskName + "-" + ((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID() which keeps the behavior as it if `transactionalPrefix` is absent.
• the transactionalPrefix value if present

transactionalIdsGenerator =
        new TransactionalIdsGenerator(
                transactionalIdPrefix != null
                    ? transactionalIdPrefix
                    : taskName
                        + "-"
                        + ((StreamingRuntimeContext) getRuntimeContext())
                                .getOperatorUniqueID(),
                ...);

Compatibility, Deprecation, and Migration Plan

This improvement is compatible and will not impact existing users, because the transactional id will keep the older behavior if it is absent.

Test Plan

It will be covered by unit and integration tests.

Rejected Alternatives

N/A