Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer by marcin-krystianc · Pull Request #4989 · confluentinc/librdkafka

Marcin Krystianc (marcin-krystianc) · 2025-03-11T09:09:09Z

Currently, the librdkafka implementation sends only one produce request at a time, regardless of the configured value for the max.in.flight.requests.per.connection parameter. This limitation prevents the producer from fully leveraging the potential of concurrent requests, which can impact throughput and performance.

With the proposed changes, the max.in.flight.requests.per.connection parameter will be respected, allowing the number of concurrent requests to be aligned with the specified value. This enhancement aims to optimize the idempotent producer's performance by enabling multiple produce requests to be sent simultaneously, thereby improving overall latency.

To validate my findings, I conducted tests on a network with a simulated delay of 1000 ms, configured using the tc command (tc qdisc add dev eth0 root netem delay 1000ms). I have also attached pcap files for your review.

Relevant producer parameters:

--config enable.idempotence=true
--config max.in.flight.requests.per.connection=5
--config linger.ms=200
--config socket.nagle.disable=true

librdkafka 2.8.0 (p95 latency ~=1950ms), capture_2.8.0.pcap:

418	13:01:07.506672	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request
432	13:01:08.508093	172.18.0.3	172.18.0.4	19092,59686	Kafka Produce v10 Response
454	13:01:08.508619	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request
459	13:01:09.509790	172.18.0.3	172.18.0.4	19092,59686	Kafka Produce v10 Response
467	13:01:09.510013	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request
474	13:01:10.511038	172.18.0.3	172.18.0.4	19092,59686	Kafka Produce v10 Response
482	13:01:10.511223	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request
487	13:01:11.512288	172.18.0.3	172.18.0.4	19092,59686	Kafka Produce v10 Response
495	13:01:11.512462	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request
500	13:01:12.513293	172.18.0.3	172.18.0.4	19092,59686	Kafka Produce v10 Response
508	13:01:12.513462	172.18.0.4	172.18.0.3	59686,19092	Kafka Produce v10 Request

librdkafka 2.8.0+patch (p95 latency ~= 1190ms), capture_2.8.0_patched.pcap:

237	12:58:32.288766	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
239	12:58:32.288771	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
246	12:58:32.329058	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
247	12:58:32.329061	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
250	12:58:32.788187	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
260	12:58:33.301184	172.18.0.3	172.18.0.4	19092,60190	Kafka Produce v10 Response
271	12:58:33.301766	172.18.0.4	172.18.0.3	60190,19092	Kafka Produce v10 Request
272	12:58:33.302767	172.18.0.3	172.18.0.4	19092,60190	Kafka Produce v10 Response
275	12:58:33.330077	172.18.0.3	172.18.0.4	19092,60190	Kafka Produce v10 Response
277	12:58:33.330703	172.18.0.3	172.18.0.4	19092,60190	Kafka Produce v10 Response

confluent-cla-assistant · 2025-03-11T09:09:19Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
✅ marcin-krystianc
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Marcin Krystianc (marcin-krystianc) · 2025-03-11T09:11:40Z

                /* Limit the number of in-flight requests (per partition)
                 * to the broker's sequence de-duplication window. */
-                max_requests = RD_MIN(max_requests,
-                                      RD_KAFKA_IDEMP_MAX_INFLIGHT - inflight);


The inflight variable holds the number of inflight messages (rd_atomic32_get(&rktp->rktp_msgs_inflight)) instead of the number of inflight requests.

Thanks Marcin Krystianc (@marcin-krystianc) for spotting this! I'll check if other changes are needed but it doesn't seem at the moment.

- CI/CD script (See #3) - confluentinc#4989 - confluentinc#4972 - confluentinc#4905 - confluentinc#4864

- confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5055 (Add missing wrlock around rd_kafka_metadata_cache_hint call)

- G-Research#3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer)

* Hotfix Release: v2.11.0-gr - collective changes - #3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) * Fix line endings * Make style checks CI job work It will fail because of some style issues from upstream, but at least it should complete instead of hanging forever. * Build for arm64 linux without emulation

Marcin Krystianc (marcin-krystianc) · 2025-08-05T15:21:27Z

Hi Emanuele Sabellico (@emasab) , do you have a rough timeline for when you anticipate being able to merge this change?

- G-Research#3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

- G-Research#3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

- #3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

- G-Research#3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

- #3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

Emanuele Sabellico (emasab) · 2026-01-14T17:09:45Z

                 * to the broker's sequence de-duplication window. */
-                max_requests = RD_MIN(max_requests,
-                                      RD_KAFKA_IDEMP_MAX_INFLIGHT - inflight);
+                max_requests = rkb->rkb_rk->rk_conf.max_inflight - rd_kafka_bufq_cnt(&rkb->rkb_waitresps);


Even with the fix it should still respect the max_requests variable that corresponds to queue.buffering.backpressure.threshold and defaults to 1 and when idempotency is enabled cannot be changed, see here. I think this was before max.in.flight was increased from 1 to 5 for the idempotent producer and we could give an error in case it's lower than 5 but by changing that configuration check.

At the moment it won't limit the in-flight requests to one once we have this fix, but only the output buffers to 1

Suggested change

max_requests = rkb->rkb_rk->rk_conf.max_inflight - rd_kafka_bufq_cnt(&rkb->rkb_waitresps);

max_requests = RD_MIN(max_requests,

RD_KAFKA_IDEMP_MAX_INFLIGHT - rd_kafka_bufq_cnt(&rkb->rkb_waitresps));

Thanks, I've tested locally whether there is any observable difference in network traffic after your suggestion, and I didn't spot any change (thus it still works as expected).

Emanuele Sabellico (emasab) · 2026-01-14T17:17:57Z

Thanks Marcin Krystianc (@marcin-krystianc) ! Verified again this one and I still agree with the change, I just I request a small edit.
About the fix, even if queue.buffering.backpressure.threshold is 1 by default as you outlined it's likely there are more than 1 message per buffer and in that case it stops enqueuing until all message responses are received ('rd_atomic32_get(&rktp->rktp_msgs_inflight)') , not until the are no buffers in the output queue rkb_outbufs, actually limiting inflight requests to 1 instead of 5.

Emanuele Sabellico (emasab) · 2026-01-14T17:19:22Z

/sem-approve

…he broker's sequence de-duplication window.

- G-Research#3 (CI/CD script) - confluentinc#4972 (Avoid unnecessary producer epoch bumps) - confluentinc#4989 (Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer) - confluentinc#5168 (Use system-provided cyrus-sasl/libsasl2 at runtime)

…the idempotent producer confluentinc#4989

* CI/CD script * Avoid unnecessary producer epoch bumps confluentinc#4972 * Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer confluentinc#4989 * Use system-provided cyrus-sasl/libsasl2 at runtime confluentinc#5168 * Update changelog

Emanuele Sabellico (emasab) · 2026-02-03T09:21:44Z

/sem-approve

…the idempotent producer confluentinc#4989

* CI/CD script * Avoid unnecessary producer epoch bumps confluentinc#4972 * Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer confluentinc#4989 * Use system-provided cyrus-sasl/libsasl2 at runtime confluentinc#5168 * Update changelog

…the idempotent producer confluentinc#4989

* CI/CD script * Avoid unnecessary producer epoch bumps confluentinc#4972 * Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer confluentinc#4989 * Use system-provided cyrus-sasl/libsasl2 at runtime confluentinc#5168 * Update changelog

Marcin Krystianc (marcin-krystianc) requested a review from a team as a code owner March 11, 2025 09:09

Marcin Krystianc (marcin-krystianc) commented Mar 11, 2025

View reviewed changes

Emanuele Sabellico (emasab) added the bug label Mar 12, 2025

Marcin Krystianc (marcin-krystianc) mentioned this pull request Mar 17, 2025

Hotfix Release: v2.8.0-gr - collective changes G-Research/librdkafka#4

Merged

Marcin Krystianc (marcin-krystianc) added a commit to G-Research/librdkafka that referenced this pull request Mar 17, 2025

Hotfix Release: v2.8.0-gr - collective changes

ae4d5d1

- CI/CD script (See #3) - confluentinc#4989 - confluentinc#4972 - confluentinc#4905 - confluentinc#4864

Marcin Krystianc (marcin-krystianc) mentioned this pull request Apr 29, 2025

Hotfix Release: v2.10.0-gr - collective changes G-Research/librdkafka#5

Merged

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Jul 31, 2025

Hotfix Release: v2.11.0-gr - collective changes G-Research/librdkafka#6

Merged

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Sep 8, 2025

Hotfix Release: v2.11.1.1+gr - collective changes G-Research/librdkafka#10

Merged

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Oct 9, 2025

Hotfix Release: v2.12.0.1+gr - collective changes G-Research/librdkafka#11

Merged

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Oct 22, 2025

Hotfix Release: v2.12.1.1+gr - collective changes G-Research/librdkafka#12

Merged

Emanuele Sabellico (emasab) requested changes Jan 14, 2026

View reviewed changes

Marcin Krystianc (marcin-krystianc) force-pushed the dev-20250310-inflight branch from 402cd13 to 704b784 Compare January 15, 2026 11:40

Correctly limit the number of in-flight requests (per partition) to t…

98199cc

…he broker's sequence de-duplication window.

Marcin Krystianc (marcin-krystianc) force-pushed the dev-20250310-inflight branch from 704b784 to 98199cc Compare January 15, 2026 11:46

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Jan 16, 2026

Hotfix Release: v2.13.0.1+gr - collective changes G-Research/librdkafka#13

Merged

Jonathan Giannuzzi (jgiannuzzi) added a commit to jgiannuzzi/librdkafka that referenced this pull request Jan 16, 2026

Fully utilize the max.in.flight.requests.per.connection parameter on …

0e868d4

…the idempotent producer confluentinc#4989

Jonathan Giannuzzi (jgiannuzzi) added a commit to jgiannuzzi/librdkafka that referenced this pull request Mar 6, 2026

Fully utilize the max.in.flight.requests.per.connection parameter on …

a448cf6

…the idempotent producer confluentinc#4989

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Mar 6, 2026

Hotfix Release: v2.13.2.1+gr - collective changes G-Research/librdkafka#14

Merged

Jonathan Giannuzzi (jgiannuzzi) added a commit to jgiannuzzi/librdkafka that referenced this pull request Apr 9, 2026

Fully utilize the max.in.flight.requests.per.connection parameter on …

1b56ee6

…the idempotent producer confluentinc#4989

Jonathan Giannuzzi (jgiannuzzi) mentioned this pull request Apr 9, 2026

Hotfix Release: v2.14.0.1+gr - collective changes G-Research/librdkafka#15

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer#4989

Fully utilize the max.in.flight.requests.per.connection parameter on the idempotent producer#4989
Marcin Krystianc (marcin-krystianc) wants to merge 1 commit intoconfluentinc:masterfrom
marcin-krystianc:dev-20250310-inflight

Marcin Krystianc (marcin-krystianc) commented Mar 11, 2025

Uh oh!

confluent-cla-assistant Bot commented Mar 11, 2025

Uh oh!

Marcin Krystianc (marcin-krystianc) Mar 11, 2025

Uh oh!

Emanuele Sabellico (emasab) Mar 12, 2025

Uh oh!

Marcin Krystianc (marcin-krystianc) commented Aug 5, 2025

Uh oh!

Emanuele Sabellico (emasab) Jan 14, 2026

Uh oh!

Marcin Krystianc (marcin-krystianc) Jan 15, 2026

Uh oh!

Emanuele Sabellico (emasab) commented Jan 14, 2026 •

edited

Loading

Uh oh!

Emanuele Sabellico (emasab) commented Jan 14, 2026

Uh oh!

Emanuele Sabellico (emasab) commented Feb 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	max_requests = rkb->rkb_rk->rk_conf.max_inflight - rd_kafka_bufq_cnt(&rkb->rkb_waitresps);
	max_requests = RD_MIN(max_requests,
	RD_KAFKA_IDEMP_MAX_INFLIGHT - rd_kafka_bufq_cnt(&rkb->rkb_waitresps));

Conversation

Marcin Krystianc (marcin-krystianc) commented Mar 11, 2025

Uh oh!

confluent-cla-assistant Bot commented Mar 11, 2025

Uh oh!

Marcin Krystianc (marcin-krystianc) Mar 11, 2025

Choose a reason for hiding this comment

Uh oh!

Emanuele Sabellico (emasab) Mar 12, 2025

Choose a reason for hiding this comment

Uh oh!

Marcin Krystianc (marcin-krystianc) commented Aug 5, 2025

Uh oh!

Emanuele Sabellico (emasab) Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Marcin Krystianc (marcin-krystianc) Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

Emanuele Sabellico (emasab) commented Jan 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Emanuele Sabellico (emasab) commented Jan 14, 2026

Uh oh!

Emanuele Sabellico (emasab) commented Feb 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Emanuele Sabellico (emasab) commented Jan 14, 2026 •

edited

Loading