Status

Current state:  Discussing Voting in progress

Discussion thread: https://lists.apache.org/thread/vkoqpq10xmvvzknfn65r394d577yogrk

Vote thread: https://lists.apache.org/thread/dko0pykc8wdbwp0mzs2h2jj2zpg9vpty

JIRA:

...

./bin/kafka-producer-perf-test.sh --topic quickstart-events --num-records 10000000 --throughput 1000000000 --record-size 1000 --transactional-id thisistest --bootstrap-server 127.0.0.1:9092 --command-property max.in.flight.requests.per.connection=x

Public Interfaces

...

Topic Configuration

• Config name: max.idempotenceproducer.state.batches.to.retain
• Server default config name: log.producer.state.batches.to.retain
• Config type: Int
• Default value: 5
• Constraint: at least 5

Introduce a new config on the broker, as the broker must know how much memory to allocate. Operators can set a limitation on the broker side to prevent malicious producerstopic configuration producer.state.batches.to.retain, with a server default config log.producer.state.batches.to.retain. This configuration controls how many produce request metadata entries the broker retains per partition to support idempotent and transactional deduplication. Operators can set the server default on the broker to apply a cluster-wide baseline, or override it per topic to tune individual topics independently. This configuration only takes effect for idempotent/transactional producers. Prior to this KIP, idempotent/transactional producers are limited to a maximum of 5 for max.in.flight.requests.per.connection. Setting a value greater than 5 on these older producers causes a ConfigException at initialization. To ensure backward compatibility, the new broker configuration max.idempotenceproducer.state.batches.to.retain must have a minimum constraint of 5. This preserves the behavior of existing producers, which assume the broker always retains at least 5 batch metadata entries. 

This KIP preserves the existing default of 5 for both maxproducer.idempotencestate.batches.to.retain and max.in.flight.requests.per.connection. While higher defaults could improve throughput, such a change is not the goal of this KIP. We defer this discussion to a future proposal.

Producer Configuration

max.in.flight.requests.per.connection

Prior to this KIP, the Kafka producer client enforces a hard limit of 5 on max.in.flight.requests.per.connection when idempotence or transactions are enabled, because the broker hardcodes its deduplication window to 5 entries. Setting a higher value causes a ConfigException at producer initialization. This KIP removes that restriction. With the broker's deduplication window now configurable and discoverable per partition via ProduceResponse, the producer no longer needs a static upper bound enforced at startup. The max.in.flight.requests.per.connection reverts to its original role: a pure connection-level capacity on the total number of in-flight requests to a broker node, regardless of whether the producer is idempotent or transactional. The per-partition in-flight check described below takes over responsibility for ensuring the producer does not exceed the broker's deduplication window for any individual partition.

ProduceRequest / ProduceResopnse

Add a new field MaxIdempotenceBatchesToRetain ProducerStateBatchesToRetain to reflect broker topic partition configuration maxproducer.idempotencestate.batches.to.retain in ProduceResponse.

...

 {
   "apiKey": 0,
   "type": "response",
   "name": "ProduceResponse",
   // ...
-  "validVersions": "3-13",
+  // Versions 14 adds MaxIdempotenceBatchesToRetainProducerStateBatchesToRetain as a tagged field (KIP-1269).
+  "validVersions": "3-14",
   "flexibleVersions": "9+",
    "fields": [
     // ...
     { "name": "NodeEndpointsResponses", "type": "[]NodeEndpointTopicProduceResponse", "versions": "100+", "taggedVersions": "10+", "tag": 0,
       "about": "EndpointsEach for all current-leaders enumerated in PartitionProduceResponses, with errors NOT_LEADER_OR_FOLLOWERproduce response.", "fields": [
       { "name": "NodeIdName", "type": "int32string", "versions": "10+0-12",
         "mapKey": true, "entityType": "brokerIdtopicName", "aboutmapKey": "The ID of the associated node."},
       { "nametrue, "ignorable": "Host", "type": "string", "versions": "10+",
 true,
        "about": "The node's hostname." },
       { "name": "Port", "type": "int32", "versions": "10+",
         "about": "The node's porttopic name." },
       { "name": "RackTopicId", "type": "stringuuid", "versions": "1013+", "nullableVersionsmapKey": "10+"true, "defaultignorable": "null",
         true, "about": "The rackunique oftopic theID" node},
 or null if it has not been assigned to a rack." }
-    ]}
+ // ...
  -   ]}
  +   ]},
  +    { "name": "MaxIdempotenceBatchesToRetainProducerStateBatchesToRetain", "type": "int32",
  +     "versions": "14+", "taggedVersions": "14+", "tag": 1, "default"
: "5",
  +      "about": "The maximum number of idempotenceidempotent batches theto brokerretain retains in memorythe for a producer to a topic partition."
+broker." }
    ]},
    // }...
   ]
 }

Proposed Changes

Two-Level In-Flight Check

To support per-topic deduplication window sizes, the producer enforces two independent in-flight checks before sending a batch. The first is a per-partition check: for each topic partition, the producer tracks the number of batches currently in-flight to that partition and ensures it does not exceed the partition's discovered ProducerStateBatchesToRetain limit. This enforces the broker's deduplication window on a per-partition basis, so that a topic configured with a larger window does not interfere with a topic configured with a smaller one. The second is the existing per-connection check: the total number of in-flight requests to a broker node must not exceed max.in.flight.requests.per.connection. Both checks must pass before a batch is sent. The per-partition check is the new constraint introduced by this KIP; the per-connection check is unchanged from existing behavior. The per-partition limit is initially 5 and is updated dynamically as the producer receives ProduceResponse messages, as described in the Dynamic Capacity Discovery section below.

Dynamic Capacity Discovery

Initial State (Safety First): When a connection is first established, the

...

per-partition in-flight limit defaults to 5 for all partitions on that node. This ensures

...

safe behavior with any broker version

...

, since old brokers hardcode their deduplication window to 5 entries. The per-connection limit remains max.in.flight.requests.per.connection

...

as configured by the user.

Discovery: Upon receiving

...

a ProduceResponse from a partition leader, the producer

...

reads the ProducerStateBatchesToRetain tagged field from each PartitionProduceResponse. This value reflects the producer.state.batches.to.retain configuration of that specific topic partition on the leader broker.

Adaptation

...

:

• If the field is present, the Producer updates its local limitation for that specific node connection to min(max.in.flight.requests.per.connection, MaxIdempotenceBatchesToRetain).If the field is missing (older broker), the Producer uses limitation as initial stateproducer updates the per-partition in-flight limit for that partition to the received ProducerStateBatchesToRetain value.
• If the field is absent, the producer retains the default per-partition limit of 5 for all partitions on that node.

Leader Change: When the producer detects a partition leader change, the producer resets the per-partition in-flight limit for that partition back to 5 and re-enters the discovery process with the new leader. This is necessary because the new leader may be an old broker that does not recognize producer.state.batches.to.retain and hardcodes its deduplication window to 5. Resetting to 5 on every leader change ensures correctness regardless of whether the new leader is an upgraded or non-upgraded broker, and the producer will rediscover the correct limit from the new leader's first ProduceResponse.

Compatibility, Deprecation, and Migration Plan

New topic configuration with old broker

The producer.state.batches.to.retain has no effect on old brokers. Old brokers do not recognize this topic configuration and will continue to use their hardcoded deduplication window of 5. Even if an operator sets producer.state.batches.to.retain to a value greater than 5 on a topic, the old broker ignores it and the ProducerStateBatchesToRetain field will remain default value in PartitionProduceResponse. The new producer will therefore fall back to the default per-partition in-flight limit of 5 for all partitions on that node, as described in the Dynamic Capacity Discovery section. Operators should ensure all brokers in the cluster are upgraded before relying on values greater than 5.

New broker with old producer

The

...

minimum value of

...

producer.

...

state.batches.to.retain is 5

...

, which matches the deduplication window old producers assume. Old producers are still subject to the existing ConfigException at initialization if max.in.flight.requests.per.connection

...

is set greater than 5 with idempotence or transactions enabled. Their behavior on the broker side is unchanged.

Old broker with new producer

When a new producer connects to an old broker, the old broker doesn't set the ProducerStateBatchesToRetain in ProduceResponse. The default value of MaxIdempotenceBatchesToRetain is 5, so it works as same constraint in old brokersthe new producer retains the default per-partition in-flight limit of 5 for all partitions on that node, which matches the old broker's hardcoded deduplication window. The removal of the startup ConfigException for idempotent producers still takes effect on the producer side, but since the discovered per-partition limit defaults to 5, the producer will not send more than 5 in-flight batches per partition to an old broker, preserving correctness.

Test Plan

• Unit test

  • Validate new configuration - default/minimal value.

  • Validate producer can learn MaxIdempotenceBatchesToRetain ProducerStateBatchesToRetain from ProduceResponse and align with the brokerpartition leader.

  • Validate that the producer resets the per-partition in-flight limit to 5 upon detecting a partition leader change, and rediscovers the limit from the new leader's first ProduceResponse.
  • Validate that idempotent and transactional producers no longer throw ConfigException when max.in.flight.requests.per.connection is set greater than 5.

• Integration test

  • Producer with max.in.flight.requests.per.connection=1 and broker topic with maxproducer.idempotencestate.batches.to.retain=20: producer sends at most 1 in-flight request per partition, bounded by the connection-level check.
  • Producer with max.in.flight.requests.per.connection=1020 and broker topic with maxproducer.idempotencestate.batches.to.retain=5: producer still sends at most 5 in-flight requestbatches per partition, bounded by the per-partition check.
  • Producer with max.in.flight.requests.per.connection=20 and two topics on the same broker, one with producer.state.batches.to.retain=5 and another with producer.state.batches.to.retain=20: producer respects each partition's limit independently without one topic affecting the other.
  • Producer connected to an old broker with producer.state.batches.to.retain=20 set on the topic: producer falls back to a per-partition limit of 5 since the old broker does not return ProducerStateBatchesToRetain in PartitionProduceResponse.
  • Upon partition leader change from a new broker to an old broker: producer resets the per-partition limit to 5 and does not exceed the old broker's hardcoded deduplication window.

Rejected Alternatives

Hardcoded Increase (e.g., to 20)

• Proposal: Just bump the hardcoded NUM_BATCHES_TO_RETAIN to 20 in the broker code.

• Rejection Reason: While memory usage is generally low, Kafka is used in diverse environments. Forcing a 4x increase in per-producer memory overhead for all users—including those who don't need high BDP throughput—is not acceptable. It must be opt-in/configurable.

Return

...

ProducerStateBatchesToRetain via InitProducerIdResponse

• Proposal: Return the limit in the InitProducerIdResponse instead of ProduceResponse.

• Rejection Reason: Different brokers can have different maxproducer.idempotencestate.batches.to.retain. After getting a PID, the producer can communicate with various partition leader. In this design, the producer doesn't have chance to learn the limitation on different brokers.

...