CORE-13396 pass client configuration to consumer #27703
Open
+564
−459
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
mmaslankaprv
force-pushed
the
CORE-13396-pass-client-configuration-to-consumer
branch
from
25a9f82 to
24d565f
Compare
mmaslankaprv
changed the title
After talking to perf team we decided to adjust default parameters of consumer used to fetch data from source cluster. The parameters we use current are following: ``` fetch_max_wait: 500ms fetch_min_bytes: 5 MiB fetch_max_bytes: 20 MiB partition_max_bytes: 1 MiB ``` Signed-off-by: Michał Maślanka <[email protected]>
Added configuration allowing users to set partition max bytes Signed-off-by: Michał Maślanka <[email protected]>
Previously client configuration did not include max partition bytes. Added missing value to configure max number of bytes fetched per partition. Signed-off-by: Michał Maślanka <[email protected]>
Cleaned up creation of consumer configuration. Now all the relevant parameters can be configured through APIs. Signed-off-by: Michał Maślanka <[email protected]>
mmaslankaprv
force-pushed
the
CORE-13396-pass-client-configuration-to-consumer
branch
from
0caf813 to
a2559f6
Compare
mmaslankaprv
requested review from
Copilot,
bharathv,
oleiman,
michael-redpanda and
joe-redpanda
There was a problem hiding this comment.
Pull Request Overview
This PR passes client configuration parameters to the shadow linking Kafka consumer by adding the fetch_partition_max_bytes field and updating default values for fetch configuration. The changes enable better tuning of the underlying Kafka client performance.
Key Changes:
- Added
fetch_partition_max_bytesfield to shadow link client configuration - Updated default values for fetch timeouts and byte limits to more performant values
- Refactored implementation classes to be consolidated into service.cc
Reviewed Changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
tests/rptest/tests/cluster_linking_e2e_test.py |
Added test coverage for new fetch_partition_max_bytes configuration |
tests/rptest/clients/admin/proto/redpanda/core/admin/v2/shadow_link_pb2.* |
Generated protobuf files with new field and updated defaults |
src/v/redpanda/admin/services/shadow_link/converter.cc |
Added conversion logic for fetch_partition_max_bytes parameter |
src/v/cluster_link/tests/partition_replicator_fixture_tests.cc |
Updated test to use new API structure |
src/v/cluster_link/service.* |
Consolidated implementation classes and added configuration update support |
src/v/cluster_link/replication/* |
Added configuration update methods and removed separate implementation files |
src/v/cluster_link/model/types.h |
Added fetch_partition_max_bytes field and updated default values |
src/v/cluster_link/manager.* |
Added callback system for configuration change notifications |
proto/redpanda/core/admin/v2/shadow_link.proto |
Added fetch_partition_max_bytes field to protobuf definition |
mmaslankaprv
force-pushed
the
CORE-13396-pass-client-configuration-to-consumer
branch
from
717c1d3 to
4d0faf5
Compare
Moved source and sync implementation out of replication module. This way the `replication` sub-module does not have to depend on anything from cluster_link module and have no information about the specific implementation of actual source and sink that are used in cluster linking. Signed-off-by: Michał Maślanka <[email protected]>
Some components may have to react when link configuration changes to update its properties. Added a notification mechanism that allows registering a callback to be notified about link configuration changes. Signed-off-by: Michał Maślanka <[email protected]>
mmaslankaprv
force-pushed
the
CORE-13396-pass-client-configuration-to-consumer
branch
from
4d0faf5 to
66fc2b9
Compare
Previously direct consumer instance was injected into mux_consumer. This made the mux consumer unaware of direct consumer configuration. Changed the way how mux consumer is initialized. Delegated responsibility of instantiating direct consumer to `mux_consumer` internals. This way an ownership is clear and we can encapsulate configuration changes. Signed-off-by: Michał Maślanka <[email protected]>
Signed-off-by: Michał Maślanka <[email protected]>
When link configuration changes we must update configuration of consumer used to fetch data from the source cluster. This commits register manager notification that listens for the link configuration changes and updates the mux consumer configuration. Signed-off-by: Michał Maślanka <[email protected]>
Signed-off-by: Michał Maślanka <[email protected]>
mmaslankaprv
force-pushed
the
CORE-13396-pass-client-configuration-to-consumer
branch
from
66fc2b9 to
38f3ecb
Compare
bharathv
reviewed
Sep 25, 2025
| , _sem(max_buffered_bytes, "partition_data_queue") {} | ||
|
|
||
| void partition_data_queue::update_max_buffered(size_t new_value) { | ||
| _gate.check(); |
There was a problem hiding this comment.
I think callbacks shouldn't throw exceptions? return gracefully maybe?
| } else { | ||
| _sem.consume(_current_max_buffered - new_value); | ||
| } | ||
| _current_max_buffered = new_value; |
There was a problem hiding this comment.
nit: I think "current_max_buffered" gives an impression that that much is currently buffered, just call it "_max_buffered_bytes" or something?
Retry command for Build#72959please wait until all jobs are finished before running the slash command |
CI test resultstest results on build#72959
|
oleiman
reviewed
Sep 25, 2025
| static constexpr auto retry_backoff_ms_default = 100; | ||
| // Maximum fetch wait time | ||
| std::optional<int32_t> fetch_wait_max_ms; | ||
| // Default value for fetch_wait_max_ms (100ms) |
There was a problem hiding this comment.
nitpick: missed a spot
Suggested change
| // Default value for fetch_wait_max_ms (100ms) | |
| // Default value for fetch_wait_max_ms (500ms) |
another one below (fetch_max_bytes_default). incidentally, why bother duplicating these value in code comments at all?
Comment on lines
+382
to
+385
| assert ( | ||
| updated_link.configurations.client_options | ||
| == shadow_link.configurations.client_options | ||
| ) |
| , _sem(max_buffered_bytes, "partition_data_queue") {} | ||
|
|
||
| void partition_data_queue::update_max_buffered(size_t new_value) { | ||
| _gate.check(); |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Passing all the client configurations to shadow linking underlying kafka client.
Backports Required
Release Notes