...

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 producers. 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 initializationtopic 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. 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.

ProduceRequest / ProduceResopnse

...

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 ProducerStateBatchesToRetain to reflect topic partition configuration producer.state.batches.to.retain in ProduceResponse.

 {
   "apiKey": 0,
   "type": "request",
   "listeners": ["broker"],
   "name": "ProduceRequest",
   // ...
-  "validVersions": "3-13",
+  // Version 14 is the same as version 13 (KIP-1269).
+  "validVersions": "3-14",
   // ...
 }

 {
   "apiKey": 0,
   "type": "response",
   "name": "ProduceResponse",
   // ...
-  "validVersions": "3-13",
+  // Versions 14 adds ProducerStateBatchesToRetain as a tagged field (KIP-1269).
+  "validVersions": "3-14",
   "flexibleVersions": "9+",
   "fields": [

 {
   "apiKey": 0,
   "type": "request",
   "listeners": ["broker"],
   "name": "ProduceRequest",
   // ...
-  "validVersions": "3-13",
+  // Version 14 is the same as version 13 (KIP-1269).
+  "validVersions": "3-14",
   // ...
 }

 {
   "apiKey": 0,
   "type": "response",
   "name": "ProduceResponse",
   // ...
-  "validVersions": "3-13",
+  // Versions 14 adds MaxIdempotenceBatchesToRetain as a tagged field (KIP-1269).
+  "validVersions": "3-14",
   "flexibleVersions": "9+",
   "fields": [
     // ...
     { "name": "NodeEndpoints", "type": "[]NodeEndpoint", "versions": "10+", "taggedVersions": "10+", "tag": 0,
       "about": "Endpoints for all current-leaders enumerated in PartitionProduceResponses, with errors NOT_LEADER_OR_FOLLOWER.", "fields": [
       { "name": "NodeIdResponses", "type": "int32[]TopicProduceResponse", "versions": "100+",
         "mapKeyabout": true, "entityType": "brokerId"Each produce response.", "aboutfields": "The ID of the associated node."},
 [
      { "name": "HostName", "type": "string", "versions": "10+0-12",
         "about"entityType": "topicName", "mapKey": true, "The node's hostnameignorable": true,
        "about": "The topic name." },
       { "name": "PortTopicId", "type": "int32uuid", "versions": "1013+",
 "mapKey": true,      "ignorable": true, "about": "The unique node's port.topic ID" },
       { "name": "RackPartitionResponses", "type": "string[]PartitionProduceResponse", "versions": "100+",
 "nullableVersions       "about": "10+Each partition that we produced to within the topic.", "defaultfields": "null",[
        { "aboutname": "Index"The rack of the node, or null if it has not been assigned to a rack." }
-    ]}
+    ]},
+, "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "MaxIdempotenceBatchesToRetainErrorCode", "type": "int32int16", "versions": "140+",
 "taggedVersions": "14+", "tag": 1, "default"
: "5",
+      "about": "The maximumerror numbercode, ofor idempotence0 batchesif thethere brokerwas retainsno in memory for a producer to a topic partition."
+error." },
        { "name": "BaseOffset", "type": "int64", "versions": "0+",
      }
   ]
 }

Proposed Changes

Dynamic Capacity Discovery

Initial State (Safety First): When a connection is first established, the produce sends at most min(5, max.in.flight.requests.per.connection) for each node. This ensures safety with any broker version. Old brokers hardcode NUM_BATCHES_TO_RETAIN to 5. To ensure compatibility, the producer initially limits in-flight requests to min(5, max.in.flight.requests.per.connection) until it discovers the broker's actual capability via the ProduceResponse. This prevents sending excessive in-flight requests to older brokers that cannot handle them.

Discovery: Upon receiving the first ProduceResponse from a partition leader, the producer checks for the presence of the MaxIdempotenceBatchesToRetain tagged field.

Adaption:

 "about": "The base offset." },
        { "name": "LogAppendTimeMs", "type": "int64", "versions": "2+", "default": "-1", "ignorable": true,
          "about": "The timestamp returned by broker after appending the messages. If CreateTime is used for the topic, the timestamp will be -1.  If LogAppendTime is used for the topic, the timestamp will be the broker local time when the messages are appended." },
        { "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true,
          "about": "The log start offset." },
  +     { "name": "ProducerStateBatchesToRetain", "type": "int32",
  +       "versions": "14+", "taggedVersions": "14+", "tag": 1, "default": "5",
  +       "about": "The maximum number of idempotent batches to retain in the 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 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 is 5, so the 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 ProducerStateBatchesToRetain from ProduceResponse and align with the partition 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
  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 state.

Compatibility, Deprecation, and Migration Plan

New broker with old producer

...

• • =1 and topic with producer.state.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=20 and topic with producer.state.batches.to.retain=5: producer sends at most 5 in-flight batches per partition, bounded by the per-partition check.

Old broker with new producer

The default value of MaxIdempotenceBatchesToRetain is 5, so it works as same constraint in old brokers.

Test Plan

• Unit test

  • Validate new configuration.

  • Validate producer can learn MaxIdempotenceBatchesToRetain from ProduceResponse and align with the broker.

• Integration test
  • Producer with max.in.flight.requests.per.connection=1 and broker with max.idempotence.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 sends at most 1 in-flight request.Producer with max.in.flight.requests.per.connection=10 and broker with max.idempotence.batches.to.retain=5: producer still sends at most 5 in-flight requestfalls 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.

...