Status

Current state: Draft

Discussion thread: 

...

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

Motivation

Currently, when a Kafka producer is configured for idempotence (enable.idempotence=true), the configuration max.in.flight.requests.per.connection should be less than or equal to 5. This limitation exists because the Kafka broker hardcodes the number of distinct batch sequence numbers it retains in memory for deduplication (ProducerStateEntry#NUM_BATCHES_TO_RETAIN).

...

We propose making the broker-side deduplication buffer size configurable (or dynamic). This will allow operators to tune their clusters for high-latency environments by increasing the allowed max.in.flight.requests.per.connection for idempotent producers. This change will enable users to maintain exactly-once semantics and strict ordering without accepting a massive degradation in throughput over long-distance links.

Public Interfaces

Broker Configuration

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

• Config name: max.idempotence.batches.to.retain

• Config type: Int

• Default value: 5

• Constraint: at least 5

ProduceRequest / ProduceResopnse

Add a new field MaxIdempotenceBatchesToRetain to reflect broker configuration max.idempotence.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-12681269).
+  "validVersions": "3-14",
   // ...
 }

...

 {
   "apiKey": 0,
   "type": "response",
   "name": "ProduceResponse",
   // ...
-  "validVersions": "3-13",
+  // Versions 14 adds MaxInFlightRequestsMaxIdempotenceBatchesToRetain as a tagged field (KIP-xxx1269).
+  "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": "NodeId", "type": "int32", "versions": "10+",
         "mapKey": true, "entityType": "brokerId", "about": "The ID of the associated node."},
       { "name": "Host", "type": "string", "versions": "10+",
         "about": "The node's hostname." },
       { "name": "Port", "type": "int32", "versions": "10+",
         "about": "The node's port." },
       { "name": "Rack", "type": "string", "versions": "10+", "nullableVersions": "10+", "default": "null",
         "about": "The rack of the node, or null if it has not been assigned to a rack." }
-    ]}
+    ]},
+    { "name": "MaxIdempotenceBatchesToRetain", "type": "int32", "versions": "14+", "taggedVersions": "14+", "tag": 1, "default"
: "5",
+      "about": "The maximum number of idempotence batches the broker retains in memory for a producer to a topic partition."
+    }
   ]
 }

Proposed Changes

Dynamic Capacity Discovery

Initial State (Safety First): When a connection is first established, the produce sends at most max(5, max.in.flight.requests.per.connection) for each node. This ensures safety with any broker version.

...

• 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

The minimal value of max.idempotence.batches.to.retain is 5. Even if the old producer sets max.in.flight.requests.per.connection up to 5, it doesn’t break the broker configuration.

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.batches.to.retain=20: 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 request.

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 MaxIdempotenceBatchesToRetain via InitProducerIdResponse

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

• Rejection Reason: Different brokers can have different max.idempotence.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.

...