This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Current state: "Under Discussion"
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
KafkaProducer is designed to be thread-safe and we encourage users to share a single producer instance across multiple threads. This approach is effective for scenarios where numerous threads append records infrequently and it helps to avoid creating many producers on resource-limited systems. These threads often need to send records to different topics, which is supported by the current producer implementation. However, another important configuration—acks—is currently set at the producer level only.
This limitation prevents users from reusing KafkaProducer instance when different partitions require distinct asks settings, forcing them to create separate producers for each configuration. Furthermore, this capability has always been available at the RPC-level, and with this update, we're simply bringing it back into users' hands.
We only add a new field into ProducerRecord constructor.
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers, Short acks) {
if (topic == null)
throw new IllegalArgumentException("Topic cannot be null.");
if (timestamp != null && timestamp < 0)
throw new IllegalArgumentException(
String.format("Invalid timestamp: %d. Timestamp should always be non-negative or null.", timestamp));
if (partition != null && partition < 0)
throw new IllegalArgumentException(
String.format("Invalid partition: %d. Partition number should always be non-negative or null.", partition));
this.topic = topic;
this.partition = partition;
this.key = key;
this.value = value;
this.timestamp = timestamp;
this.headers = new RecordHeaders(headers);
this.acks = acks; // new field
} |