[CORE-12817] - Hook kafka::client::cluster into link #26857

michael-redpanda · 2025-07-16T19:22:14Z

This PR introduces the cluster_link::remote_cluster_connection abstract class that is used by cluster_link::link and relevant tasks to communicate with the source cluster.

Backports Required

Release Notes

None

michael-redpanda · 2025-07-16T20:31:09Z

Force push:

Added client_id to model config

vbotbuildovich · 2025-07-17T00:12:38Z

CI test results

test results on build#69127

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
DataMigrationsApiTest	test_creating_and_listing_migrations		ducktape	https://buildkite.com/redpanda/redpanda/builds/69127#0198151f-1371-4f06-90bf-dc7c4d2c7035	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS
DatalakeE2ETests	test_json_schema_unicode	{"catalog_type": "rest_hadoop", "cloud_storage_type": 1, "query_engine": "trino"}	ducktape	https://buildkite.com/redpanda/redpanda/builds/69127#0198151f-1371-4f06-90bf-dc7c4d2c7035	FLAKY	19/21	upstream reliability is '98.18181818181819'. current run reliability is '90.47619047619048'. drift is 7.70563 and the allowed drift is set to 50. The test should PASS
RandomNodeOperationsTest	test_node_operations	{"cloud_storage_type": 1, "compaction_mode": "sliding_window", "enable_failures": false, "mixed_versions": false, "with_iceberg": false}	ducktape	https://buildkite.com/redpanda/redpanda/builds/69127#01981520-4813-476f-9434-221ee3d1efe4	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

test results on build#69201

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
MasterTestSuite	test_replica_pair_frequency		unit	https://buildkite.com/redpanda/redpanda/builds/69201#0198194c-43d5-4769-86ab-de9cffbc14cb	FAIL	0/1
DataMigrationsApiTest	test_creating_and_listing_migrations		ducktape	https://buildkite.com/redpanda/redpanda/builds/69201#01981968-73cb-41e3-a72b-f2971405bfe8	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

test results on build#69351

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
NodePoolMigrationTest	test_migrating_redpanda_nodes_to_new_pool	{"balancing_mode": "node_add", "cleanup_policy": "compact,delete", "test_mode": "tiered_storage"}	ducktape	https://buildkite.com/redpanda/redpanda/builds/69351#01982d99-c01f-4bb4-b4c3-9608ef03e748	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

michael-redpanda · 2025-07-17T13:42:28Z

Force push:

Rebased off dev to fix merge conflicts

michael-redpanda · 2025-07-17T14:36:22Z

Force push:

Fixed rebase snafu

oleiman

all looks pretty sensible to me. handful of nits and questions.

oleiman · 2025-07-17T17:40:04Z

src/v/kafka/client/test/cluster_mock.h

+};
+
+struct topic_metadata {
+    int32_t authorized_operations


nitpick: i agree with the bot. i had to look this up.

src/v/cluster/cluster_link/frontend.cc

src/v/cluster/cluster_link/tests/frontend_validation_test.cc

src/v/cluster_link/deps.cc

oleiman · 2025-07-17T22:34:23Z

src/v/cluster_link/deps.cc

+                  vassert(
+                    std::holds_alternative<model::tls_file_path>(
+                      link_connection.key.value()),


^^ this (Copilot's suggestion) is phrased stupidly, but it might be a bit neater w/ a visitor that takes multiple variant instances

like this:

struct visitor { auto operator()(model::tls_file_path&, model::tls_file_path&) { // OK! return stuff; } .... template<typename T1, typename T2> requires(!std::is_same_v<T1, T2>) auto operator()(T1&, T2&) { vassert(false, "OH NO!"); } };

then you can call it like

auto my_stuff = std::visit(visitor{}, connection.cert.value(), connection.key.value());

oleiman · 2025-07-17T22:35:58Z

src/v/cluster_link/deps.cc

+                  vassert(
+                    std::holds_alternative<model::tls_file_path>(
+                      link_connection.key.value()),


my other question is whether this should actually assert. where is the sameness of the key & cert variant types enforced, and if violated does that indicate a logic error in the code?

src/v/cluster_link/manager.cc

src/v/kafka/client/topic_cache.h

mmaslankaprv · 2025-07-18T06:48:01Z

src/v/kafka/client/topic_cache.h

@@ -49,6 +52,13 @@ class topic_cache {
    std::optional<kafka::leader_epoch>
      leader_epoch(model::topic_partition_view) const;

+    chunked_vector<model::topic> topics() const;


why do we need a copy ? It would be better to return topics_t const ref

Also copilot suggest using view, it is also an option here

mmaslankaprv · 2025-07-18T06:54:02Z

src/v/cluster_link/model/types.h

+    /// The CA file to use
+    std::optional<tls_file_or_value> ca;
+    /// The client ID to use
+    ss::sstring client_id;


This could be autogenerated by the cluster linking subsystem f.e. based on target cluster id. This way we will be able to identify the client logs related with a certain link

I was thinking of using the cluster link name as part of the client id

mmaslankaprv · 2025-07-18T06:54:52Z

src/v/cluster_link/model/types.h

@@ -105,6 +105,11 @@ struct scram_credentials
    auto serde_fields() { return std::tie(username, password, mechanism); }
 };

+using tls_file_path = named_type<ss::sstring, struct tls_file_path_tag>;


do we even need paths ?

I guess depends on how the customer wants to deploy the solution... the option is there

src/v/cluster_link/deps.h

mmaslankaprv

overall looks good, left some comments

michael-redpanda · 2025-07-18T12:04:10Z

@mmaslankaprv when I was working on my next task, I discovered that you already kind of built in a 'mock' system within cluster that allows us to provide 'mocked' brokers. I think that's a better system than the remote_cluster_connection I did in here. I'm going to make some changes to use that instead.

src/v/cluster/cluster_link/frontend.cc

bharathv · 2025-07-18T21:05:34Z

src/v/cluster/cluster_link/frontend.cc

+  cluster::cluster_link::errc error_code) const {
+    if (config.bootstrap_servers.empty()) {
+        vlog(
+          cluster::clusterlog.info,


bharathv · 2025-07-18T21:08:56Z

src/v/cluster_link/deps.h

+    virtual ~link_factory() = default;
+
+    virtual std::unique_ptr<link> create_link(
+      ::model::node_id self,


nit: curious why ::model..

Because there's a cluster_link::model namespace

bharathv · 2025-07-18T21:12:22Z

src/v/cluster_link/deps.cc

+        if (res.code() != errc::success) {
+            co_await handle_exception([this] { return _cluster.stop(); });
+        }


can we ensure that anything that is start()-ed should be stop()-ed too (will avoid these patterns).

bharathv · 2025-07-18T21:15:29Z

src/v/cluster_link/deps.cc

+
+        try {
+            auto describe_configs_version
+              = co_await _cluster.supported_api_versions(


whats the retry story? This may be flaky even if one broker doesn't respond.

Copilot

Pull Request Overview

This PR introduces infrastructure to hook the kafka::client::cluster component into the cluster link system, enabling tasks to communicate with remote clusters. The changes include extending the Kafka client's topic cache with new metadata fields, creating utility functions to convert cluster link metadata to Kafka client configuration, and updating the link and task systems to pass cluster connections through the dependency chain.

Extends Kafka client topic cache to track authorized operations and provide new accessor methods
Adds utility functions to convert cluster link metadata to Kafka client configuration with TLS and SASL support
Updates link and task creation to include cluster connection dependencies
Adds comprehensive test coverage for the new utilities and integration points

Reviewed Changes

Copilot reviewed 37 out of 37 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/v/kafka/client/topic_cache.h	Adds authorized_operations field and new accessor methods
src/v/kafka/client/topic_cache.cc	Implements new topic cache methods for operations and partition counts
src/v/kafka/client/test/direct_consumer_test.cc	Updates test mocks to use new topic_data structure
src/v/kafka/client/test/cluster_test_with_mock.cc	Adds test for topic metadata functionality
src/v/kafka/client/test/cluster_mock.h	Extends mock to support topic metadata with authorized operations
src/v/kafka/client/test/cluster_mock.cc	Implements topic metadata support in cluster mock
src/v/kafka/client/cluster.cc	Enables topic authorized operations in metadata requests
src/v/kafka/client/BUILD	Makes cluster library publicly visible
src/v/cluster_link/utils.h	Declares utility for converting metadata to Kafka config
src/v/cluster_link/utils.cc	Implements metadata to Kafka config conversion with TLS/SASL support
src/v/cluster_link/tests/utils_test.cc	Comprehensive tests for utility functions
src/v/cluster_link/task.h	Updates task constructor to accept link pointer
src/v/cluster_link/task.cc	Implements link pointer storage in task base class
src/v/cluster_link/link.h	Updates link to store cluster connection
src/v/cluster_link/link.cc	Implements cluster connection lifecycle management
src/v/cluster_link/manager.h	Updates manager to use cluster factory
src/v/cluster_link/manager.cc	Implements cluster factory integration
src/v/cluster_link/deps.h	Defines cluster factory and updated link factory interfaces
src/v/cluster_link/deps.cc	Implements default cluster factory
src/v/cluster_link/model/types.h	Updates connection config with new TLS field types
src/v/cluster_link/model/types.cc	Implements formatters for new TLS types

Comments suppressed due to low confidence (1)

src/v/cluster_link/tests/utils_test.cc:86

[nitpick] Using EXPECT_DEATH in unit tests can be fragile and may not work reliably in all environments. Consider testing the error condition through a more controlled mechanism.

    EXPECT_DEATH(

src/v/kafka/client/topic_cache.h

Copilot · 2025-07-19T02:03:26Z

src/v/cluster_link/utils.cc

+
+    template<typename T1, typename T2>
+    requires(!std::is_same_v<T1, T2>)
+    kafka::client::key_store operator()(const T1&, const T2&) {


[nitpick] The requires clause and vassert combination creates complex error handling. Consider using a more explicit approach with clear error types or exceptions that provide better diagnostic information.

Copilot · 2025-07-19T02:03:27Z

src/v/cluster_link/model/types.h

+        }
+
+        if (it != end && *it != '}') {
+            throw fmt::format_error("invalid format specifier for principal");


The error message mentions 'principal' but this formatter is for tls_file_or_value. The error message should be updated to reflect the correct type being formatted.

Suggested change

throw fmt::format_error("invalid format specifier for principal");

throw fmt::format_error("invalid format specifier for tls_file_or_value");

Copilot · 2025-07-19T02:03:27Z

src/v/cluster_link/model/types.h

+        }
+
+        if (it != end && *it != '}') {
+            throw fmt::format_error("invalid format specifier for principal");


The error message mentions 'principal' but this formatter is for optional<tls_file_or_value>. The error message should be updated to reflect the correct type being formatted.

Suggested change

throw fmt::format_error("invalid format specifier for principal");

throw fmt::format_error("invalid format specifier for optional<tls_file_or_value>");

Added named types for topic and cluster authorized operations field in metadata response. Signed-off-by: Michael Boquard <[email protected]>

Signed-off-by: Michael Boquard <[email protected]>

TLS settings can now be provided via a file path or the PEM value Signed-off-by: Michael Boquard <[email protected]>

Improved validation of TLS configuration Signed-off-by: Michael Boquard <[email protected]>

Signed-off-by: Michael Boquard <[email protected]>

Useful for dependencies Signed-off-by: Michael Boquard <[email protected]>

Signed-off-by: Michael Boquard <[email protected]>

Used to create instances of kafka::client::cluster Signed-off-by: Michael Boquard <[email protected]>

Signed-off-by: Michael Boquard <[email protected]>

While looping through the test factories and registering new tasks, it is possible that the task_factories would be modified while awaiting the register coroutine. Lock the task reconciler mutex while registering new task factories to ensure that doesn't happen. Signed-off-by: Michael Boquard <[email protected]>

When a task is created, give it a pointer back to the owning link so it can be referenced by the task. Will be helpful for when tasks need access to the remote_cluster_connection instance. Signed-off-by: Michael Boquard <[email protected]>

michael-redpanda requested review from bharathv, oleiman, mmaslankaprv and Copilot July 16, 2025 19:22

michael-redpanda self-assigned this Jul 16, 2025

github-actions bot added area/build area/redpanda labels Jul 16, 2025

This comment was marked as outdated.

Sign in to view

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from 878bb8e to 7897ce2 Compare July 16, 2025 20:28

michael-redpanda requested a review from Copilot July 16, 2025 20:31

This comment was marked as outdated.

Sign in to view

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from 7897ce2 to 1301ddd Compare July 17, 2025 13:42

michael-redpanda requested a review from Copilot July 17, 2025 13:42

This comment was marked as outdated.

Sign in to view

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from 1301ddd to 9010228 Compare July 17, 2025 14:31

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from 9010228 to 001c62f Compare July 17, 2025 15:53

oleiman reviewed Jul 17, 2025

View reviewed changes

mmaslankaprv reviewed Jul 18, 2025

View reviewed changes

src/v/kafka/client/topic_cache.h Outdated Show resolved Hide resolved

mmaslankaprv reviewed Jul 18, 2025

View reviewed changes

src/v/cluster_link/deps.h Outdated Show resolved Hide resolved

mmaslankaprv reviewed Jul 18, 2025

View reviewed changes

bharathv reviewed Jul 18, 2025

View reviewed changes

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from de352f3 to 4e5bf97 Compare July 19, 2025 02:01

michael-redpanda requested review from Copilot, mmaslankaprv, oleiman and bharathv July 19, 2025 02:02

Copilot AI reviewed Jul 19, 2025

View reviewed changes

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch 2 times, most recently from 135c05a to f27d2ab Compare July 20, 2025 02:13

k/protocol: Added named types for authorized ops

a84af38

Added named types for topic and cluster authorized operations field in metadata response. Signed-off-by: Michael Boquard <[email protected]>

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch 2 times, most recently from 5e73d0b to cc730c2 Compare July 21, 2025 11:55

michael-redpanda added 14 commits July 21, 2025 10:37

k/c/cluster: Tracking topic authorized operations

691f167

Signed-off-by: Michael Boquard <[email protected]>

cl/model: Updated tls settings

4df7878

TLS settings can now be provided via a file path or the PEM value Signed-off-by: Michael Boquard <[email protected]>

c/cl: Updated validation

734efdf

Improved validation of TLS configuration Signed-off-by: Michael Boquard <[email protected]>

cl/model: Added client_id to connection config

0a33a0a

Signed-off-by: Michael Boquard <[email protected]>

cl: Introduce deps

6b21749

Useful for dependencies Signed-off-by: Michael Boquard <[email protected]>

cl: Updated error codes

6150ebf

Signed-off-by: Michael Boquard <[email protected]>

k/client/bzl: Make cluster public

a83b5f2

Signed-off-by: Michael Boquard <[email protected]>

cl: Added utility to create k/client config

5187c72

Signed-off-by: Michael Boquard <[email protected]>

cl: Introduce cluster_factory

5a45927

Used to create instances of kafka::client::cluster Signed-off-by: Michael Boquard <[email protected]>

cl/tests: Wire in cluster_mock

5580c57

Signed-off-by: Michael Boquard <[email protected]>

cl: Wire in cluster_factory into manager

d47c97d

Signed-off-by: Michael Boquard <[email protected]>

cl: Wire in cluster to link

7603d9d

Signed-off-by: Michael Boquard <[email protected]>

cl: Wire in link instance to task creation

dfb67d7

When a task is created, give it a pointer back to the owning link so it can be referenced by the task. Will be helpful for when tasks need access to the remote_cluster_connection instance. Signed-off-by: Michael Boquard <[email protected]>

michael-redpanda force-pushed the dr/task/auto-topic/core-12817 branch from cc730c2 to dfb67d7 Compare July 21, 2025 14:45

	throw fmt::format_error("invalid format specifier for principal");
	throw fmt::format_error("invalid format specifier for tls_file_or_value");

[CORE-12817] - Hook kafka::client::cluster into link #26857

Are you sure you want to change the base?

[CORE-12817] - Hook kafka::client::cluster into link #26857

Uh oh!

Conversation

michael-redpanda commented Jul 16, 2025

Backports Required

Release Notes

Uh oh!

This comment was marked as outdated.

Uh oh!

michael-redpanda commented Jul 16, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

vbotbuildovich commented Jul 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI test results

Uh oh!

michael-redpanda commented Jul 17, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

michael-redpanda commented Jul 17, 2025

Uh oh!

oleiman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

oleiman Jul 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

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!

mmaslankaprv left a comment

Choose a reason for hiding this comment

Uh oh!

michael-redpanda commented Jul 18, 2025

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Copilot AI Jul 19, 2025

vbotbuildovich commented Jul 17, 2025 •

edited

Loading

oleiman Jul 17, 2025 •

edited

Loading