Configurable kafka producer parameters by SylvainSenechal · Pull Request #2789 · scality/backbeat

SylvainSenechal · 2026-08-03T14:31:29Z

Related Zenko Operator : https://github.com/scality/zenko-operator/pull/625

bert-e · 2026-08-03T14:31:33Z

Hello sylvainsenechal,

My role is to assist you with the merge of this
pull request. Please type @bert-e help to get information
on this process, or consult the user documentation.

Available options

name	description	privileged	authored
`/after_pull_request`	Wait for the given pull request id to be merged before continuing with the current one.
`/bypass_author_approval`	Bypass the pull request author's approval	⭐
`/bypass_build_status`	Bypass the build and test status	⭐
`/bypass_commit_size`	Bypass the check on the size of the changeset `TBA`	⭐
`/bypass_incompatible_branch`	Bypass the check on the source branch prefix	⭐
`/bypass_jira_check`	Bypass the Jira issue check	⭐
`/bypass_peer_approval`	Bypass the pull request peers' approval	⭐
`/bypass_leader_approval`	Bypass the pull request leaders' approval	⭐
`/approve`	Instruct Bert-E that the author has approved the pull request.		✍️
`/create_pull_requests`	Allow the creation of integration pull requests.
`/create_integration_branches`	Allow the creation of integration branches.
`/no_octopus`	Prevent Wall-E from doing any octopus merge and use multiple consecutive merge instead
`/unanimity`	Change review acceptance criteria from `one reviewer at least` to `all reviewers`
`/wait`	Instruct Bert-E not to run until further notice.

Available commands

name	description	privileged
`/help`	Print Bert-E's manual in the pull request.
`/status`	Print Bert-E's current status in the pull request `TBA`
`/clear`	Remove all comments from Bert-E from the history `TBA`
`/retry`	Re-start a fresh build `TBA`
`/build`	Re-start a fresh build `TBA`
`/force_reset`	Delete integration branches & pull requests, and restart merge process from the beginning.
`/reset`	Try to remove integration branches unless there are commits on them which do not appear on the source branch.

Status report is not available.

codecov · 2026-08-03T14:33:02Z

Codecov Report

❌ Patch coverage is 75.00000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 75.43%. Comparing base (99c1281) to head (38ef032).

Files with missing lines	Patch %	Lines
lib/BackbeatProducer.js	75.00%	1 Missing ⚠️

❌ Your patch check has failed because the patch coverage (75.00%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage.

Additional details and impacted files

Files with missing lines	Coverage Δ
extensions/ingestion/IngestionConfigValidator.js	`100.00% <ø> (ø)`
lib/config.joi.js	`100.00% <ø> (ø)`
lib/queuePopulator/IngestionPopulator.js	`67.11% <ø> (ø)`
lib/BackbeatProducer.js	`89.38% <75.00%> (+0.09%)`	⬆️

Components	Coverage Δ
Bucket Notification	`80.22% <ø> (ø)`
Core Library	`81.23% <75.00%> (+<0.01%)`	⬆️
Ingestion	`70.13% <ø> (ø)`
Lifecycle	`80.25% <ø> (ø)`
Oplog Populator	`85.83% <ø> (ø)`
Replication	`61.98% <ø> (ø)`
Bucket Scanner	`85.76% <ø> (ø)`

@@               Coverage Diff                @@
##           development/9.5    #2789   +/-   ##
================================================
  Coverage            75.43%   75.43%           
================================================
  Files                  201      201           
  Lines                13928    13929    +1     
================================================
+ Hits                 10507    10508    +1     
  Misses                3411     3411           
  Partials                10       10

Flag	Coverage Δ
api:retry	`9.06% <75.00%> (+<0.01%)`	⬆️
api:routes	`8.83% <75.00%> (+<0.01%)`	⬆️
bucket-scanner	`85.76% <ø> (ø)`
ft_test:queuepopulator	`9.13% <75.00%> (+<0.01%)`	⬆️
ingestion	`12.26% <75.00%> (+<0.01%)`	⬆️
lib	`8.79% <75.00%> (+<0.01%)`	⬆️
lifecycle	`19.24% <75.00%> (-0.01%)`	⬇️
notification	`1.00% <0.00%> (-0.01%)`	⬇️
oplogPopulator	`0.13% <0.00%> (-0.01%)`	⬇️
replication	`18.81% <75.00%> (+<0.01%)`	⬆️
unit	`54.74% <75.00%> (+0.08%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Issue: BB-685

claude · 2026-08-03T14:48:56Z

        site: joi.string(),
        compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
        requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
+        producerParams: joi.object().unknown(true).default({}),


kafka.producerParams is added to the global config, but only IngestionPopulator._setupProducer forwards it when creating a BackbeatProducer. Other producer call sites (LogReader._setupProducer, LifecycleQueuePopulator._setupProducer, BackbeatConsumer, GarbageCollectorProducer, etc.) pass compressionType/requiredAcks/maxRequestSize from kafkaConfig but skip producerParams — so the global setting silently has no effect on them.

Either forward producerParams in the other _setupProducer methods too, or move this config under extensions.ingestion only to avoid the misleading global scope.

This is intentional, the objective is to have config that will later be extensible

That's a good call from claude, should we define config per producer type or having a global one. What do you mean by extensible ?

The idea is we have
kafka.producerParams
extensions.ingestion.producerParams

extensions override the first one when provided. But for now we only have extension for ingestion, and aren't gonna bother doing it for all backbeat extensions.

I think its fine as is, but yeah eventually we should update all other backbeat extension to use this global kafka params, else its a bit weird @francoisferrand

Agree, can you do a follow up ?

bert-e · 2026-08-04T08:29:14Z

Waiting for approval

The following approvals are needed before I can proceed with the merge:

the author
2 peers

Copilot

Pull request overview

This PR extends Backbeat’s Kafka producer configuration to support additional, user-defined librdkafka producer parameters, enabling more flexible tuning (e.g., buffering settings) via Backbeat’s global config and the ingestion extension config.

Changes:

Add producerParams support to BackbeatProducer config validation and include it in the generated producerConfig (while keeping built-in critical params authoritative).
Merge global kafka.producerParams with ingestion extension producerParams in IngestionPopulator (extension overrides global).
Extend Joi schemas and add unit tests covering both BackbeatProducer behavior and ingestion merge semantics.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
tests/unit/ingestion/IngestionPopulator.js	Adds unit tests verifying merge/override behavior for global vs ingestion `producerParams`.
tests/unit/backbeatProducer.js	Adds unit tests verifying `producerParams` inclusion and protection of built-in producer config keys.
lib/queuePopulator/IngestionPopulator.js	Merges global and ingestion producer params when constructing the ingestion producer.
lib/config.joi.js	Adds `kafka.producerParams` to the main Backbeat config schema (default `{}`).
lib/BackbeatProducer.js	Accepts `producerParams`, stores them, and includes them in `producerConfig`.
extensions/ingestion/IngestionConfigValidator.js	Adds `producerParams` to ingestion extension config schema (default `{}`).

DarkIsDude · 2026-08-05T09:26:48Z

    processor: joi.object({
        circuitBreaker: joi.object().optional(),
    }).optional(),
+    producerParams: joi.object().unknown(true).default({}),


Why should we accept unknown ? Here we don't really define them neither validate them, why ? We should define what is behind object ?

there are dozens of potentials parameters here : https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html

So we went with this to not have to enumerate all, we are only using one or two for now, but wanna be able to use more without having to do another backbeat pr

Honestly debatable though, an ai could write down instantly the ~50 different params available 🤔

@francoisferrand Should we enumerate all params, or enumerate only the ones we use now, or keep it that way.

Suggested change

producerParams: joi.object().unknown(true).default({}),

producerParams: joi.object({

'queue.buffering.max.kbytes': joi.number().optional(),

'queue.buffering.max.messages': joi.number().optional(),

}).optional().default({}),

No enumerate all of them. We use only a few, and we can use only a few. So just them should be enough. The goal of Joi is to validate, here we just skip it. Also we can make sure for example that we don't insert critical one (I remember the host later that is skipped for example).

DarkIsDude · 2026-08-05T09:27:12Z

            requiredAcks: this.kafkaConfig.requiredAcks,
+            producerParams: {
+                ...this.kafkaConfig.producerParams,
+                ...this.ingestionConfig.producerParams, // Extension params override global params


Suggested change

...this.ingestionConfig.producerParams, // Extension params override global params

...this.ingestionConfig.producerParams,

I wanna keep this one, i know its more obvious for you but not for me 🧐

🤝 let's go then. You just describe the code but why not

DarkIsDude · 2026-08-05T09:28:49Z

                maxRequestSize: joi.number().default(KAFKA_PRODUCER_MESSAGE_MAX_BYTES),
                compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
                requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
+                producerParams: joi.object().unknown(true).default({}),


Same here, we should define them ? Maybe also factorise them between both (three with kafka params?)?

discussion ongoing in the other topic

DarkIsDude · 2026-08-05T09:32:10Z

        site: joi.string(),
        compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
        requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
+        producerParams: joi.object().unknown(true).default({}),


That's a good call from claude, should we define config per producer type or having a global one. What do you mean by extensible ?

DarkIsDude · 2026-08-05T09:33:50Z

+            assert.strictEqual(config['queue.buffering.max.messages'], 200000);
+        });
+
+        it('should not let producerParams override critical built-in params', () => {


Should it be rejected by joi directly by disallow some keys ?

Let's rediscuss after we've finalized joi config, some ongoing discussion on it

bert-e · 2026-08-05T09:34:06Z

Waiting for approval

The following approvals are needed before I can proceed with the merge:

the author
2 peers

The following reviewers are expecting changes from the author, or must review again:

@DarkIsDude

DarkIsDude

Keeping request changes. Waiting for Joi decision to finish the review 🙏

configurable kafka producer parameters

38ef032

Issue: BB-685

SylvainSenechal force-pushed the improvement/BB-685 branch from c772858 to 38ef032 Compare August 3, 2026 14:45

claude Bot reviewed Aug 3, 2026

View reviewed changes

scality deleted a comment from bert-e Aug 4, 2026

SylvainSenechal marked this pull request as ready for review August 4, 2026 13:00

SylvainSenechal requested review from a team, DarkIsDude, Copilot and delthas August 4, 2026 13:00

Copilot started reviewing on behalf of SylvainSenechal August 4, 2026 13:01 View session

Copilot AI reviewed Aug 4, 2026

View reviewed changes

DarkIsDude requested changes Aug 5, 2026

View reviewed changes

SylvainSenechal requested a review from DarkIsDude August 7, 2026 16:23

DarkIsDude requested changes Aug 10, 2026

View reviewed changes

-    producerParams: joi.object().unknown(true).default({}),
+    producerParams: joi.object({
+    'queue.buffering.max.kbytes': joi.number().optional(),
+    'queue.buffering.max.messages': joi.number().optional(),
+}).optional().default({}),

	...this.ingestionConfig.producerParams, // Extension params override global params
	...this.ingestionConfig.producerParams,

Uh oh!

Conversation

SylvainSenechal commented Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

bert-e commented Aug 3, 2026

Hello sylvainsenechal,

Uh oh!

codecov Bot commented Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

claude Bot Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bert-e commented Aug 4, 2026

Waiting for approval

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bert-e commented Aug 5, 2026

Waiting for approval

Uh oh!

DarkIsDude left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

SylvainSenechal commented Aug 3, 2026 •

edited

Loading

codecov Bot commented Aug 3, 2026 •

edited

Loading