cluster_link: e2e-test SR-sync against a Confluent source

Split the SR-sync e2e test into a non-collected base holding the shared
logic plus per-source leaves: the existing Redpanda-source test (same
test-ids) and a new Confluent-source leaf backed by an Apache Kafka
cluster. Source registry access goes through _make_source_client /
_source_sr_url hooks so the same validation surface runs against either
vendor.

secondary_num_brokers lets the Confluent leaf shrink the Kafka source to
2 brokers, fitting the inherited 6-node budget (3 dest + 2 kafka + 1 SR).

Author: mnajda-redpanda

tests/rptest/tests/cluster_linking_schema_registry_sync_test.py

@@ -27,7 +27,12 @@
 from rptest.clients.rpk import RpkTool
 from rptest.services.admin import Admin
 from rptest.services.cluster import TestContext, cluster
-from rptest.services.multi_cluster_services import SecondaryClusterArgs
+from rptest.services.confluent_schema_registry import ConfluentSchemaRegistryService
+from rptest.services.multi_cluster_services import (
+    SecondaryClusterArgs,
+    SecondaryClusterSpec,
+    ServiceType,
+)
 from rptest.services.redpanda import SchemaRegistryConfig
 from rptest.tests.cluster_linking_test_base import ShadowLinkTestBase
 from rptest.tests.schema_registry_test import SchemaRegistryRedpandaClient
@@ -38,9 +43,14 @@
 Pair = tuple[str, int]
 
 
-class SchemaRegistrySyncE2ETest(ShadowLinkTestBase):
+class SchemaRegistrySyncE2EBase(ShadowLinkTestBase):
     """Source cluster SR -> destination cluster SR over the shadow-link HTTP
-    source reader, at scale and with concurrent / post-sync additions."""
+    source reader, at scale and with concurrent / post-sync additions.
+
+    Shared implementation for the per-source-vendor leaf classes below. Ducktape
+    only collects leaf Test classes (no subclasses), so this base is not run
+    directly; the concrete `SchemaRegistrySyncE2ETest` (Redpanda source) and
+    `ConfluentSchemaRegistrySyncE2ETest` (Confluent source) leaves are."""
 
     # A layered diamond DAG (top -> mid -> leaf), like the reconciler
     # concurrent_stress unit test but larger: adjacent referrers share referents,
@@ -67,6 +77,19 @@ def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any):
             **kwargs,
         )
 
+    # --- source Schema Registry hooks --------------------------------------
+    # Overridden by cross-vendor subclasses (e.g. a Confluent Platform source)
+    # to point the seeding client and the shadow-link source URL at a
+    # non-Redpanda registry. The destination side stays Redpanda throughout.
+
+    def _make_source_client(self) -> SchemaRegistryRedpandaClient:
+        return SchemaRegistryRedpandaClient(self.source_cluster_service)
+
+    def _source_sr_url(self) -> str:
+        # Derive from the seeding client so the link source_url and the client
+        # can never point at different registries.
+        return self._make_source_client().base_uri()
+
     # --- schema builders ---------------------------------------------------
 
     def _record(self, name: str, fields: list[dict]) -> dict:
@@ -199,7 +222,7 @@ def _create_sr_link(
         # subsequent full syncs land quickly. An optional exact-subject filter
         # scopes replication. Returns the source URL used.
         if source_url is None:
-            source_url = self.source_cluster_service.schema_reg(limit=1)
+            source_url = self._source_sr_url()
         req = self.create_default_link_request(
             link_name=LINK_NAME,
             mirror_all_topics=False,
@@ -440,7 +463,7 @@ def _verify_rpk_status(self, expected_subjects: int, expected_versions: int):
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_e2e(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
 
         # The destination `_schemas` topic is NOT warmed up here on purpose:
@@ -517,7 +540,7 @@ def test_schema_registry_api_sync_e2e(self):
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_soft_delete(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
 
         # seeded-value: mixed active/deleted before the first sync.
@@ -559,7 +582,7 @@ def _v2(name: str) -> dict:
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_out_of_scope_reference(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
 
         # leaf-0 is a referent; mid-0 references it; ok-value is independent.
@@ -601,7 +624,7 @@ def errored() -> bool:
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_recovers_after_source_unavailable(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
 
         self._register(
@@ -630,7 +653,7 @@ def parked() -> bool:
 
         # Repoint the link at the real source. The config change forces a fresh
         # full sync, which now reaches the source and replicates.
-        good_url = self.source_cluster_service.schema_reg(limit=1)
+        good_url = self._source_sr_url()
         link = self.get_link(LINK_NAME)
         link.configurations.schema_registry_sync_options.shadow_schema_registry_api.source_url = good_url
         self.update_link(
@@ -659,7 +682,7 @@ def recovered() -> bool:
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_memory_backpressure(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
 
         # Shrink the reconcile memory budget to its 1 MiB floor and raise
@@ -689,7 +712,7 @@ def test_schema_registry_api_sync_memory_backpressure(self):
 
     @cluster(num_nodes=6)
     def test_schema_registry_api_sync_survives_leadership_change(self):
-        src = SchemaRegistryRedpandaClient(self.source_cluster_service)
+        src = self._make_source_client()
         dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
         admin = Admin(self.target_cluster_service)
 
@@ -836,3 +859,65 @@ def returning_leader_fresh() -> bool:
             backoff_sec=1,
             err_msg="reused instance did not reset state on regaining leadership",
         )
+
+
+class SchemaRegistrySyncE2ETest(SchemaRegistrySyncE2EBase):
+    """Redpanda source cluster variant: the base's default source hooks point at
+    the secondary Redpanda cluster's Schema Registry."""
+
+
+class ConfluentSchemaRegistrySyncE2ETest(SchemaRegistrySyncE2EBase):
+    """Runs the full SchemaRegistrySyncE2ETest validation surface with a
+    Confluent Platform Schema Registry (backed by an Apache Kafka source
+    cluster) as the shadow-link source instead of a secondary Redpanda cluster.
+    The destination stays Redpanda. See CORE-16776 / CORE-16357.
+
+    Node budget: 3 (destination Redpanda) + 2 (Apache Kafka source) + 1
+    (Confluent SR) = 6, matching the inherited @cluster(num_nodes=6).
+    """
+
+    # Two-broker Kafka source so the topology uses exactly the inherited
+    # num_nodes budget (see class docstring).
+    secondary_num_brokers = 2
+
+    def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any):
+        # Skip SchemaRegistrySyncE2ETest.__init__, which wires a Redpanda
+        # secondary SR; the source here is Apache Kafka + Confluent SR, so go
+        # straight to the shadow-link base with only the destination SR set.
+        ShadowLinkTestBase.__init__(
+            self,
+            test_context,
+            schema_registry_config=SchemaRegistryConfig(),
+            *args,
+            **kwargs,
+        )
+        self._confluent_sr: ConfluentSchemaRegistryService | None = None
+
+    def get_source_cluster_spec(self) -> SecondaryClusterSpec:
+        # Source cluster is Apache Kafka (KRaft, secondary_num_brokers nodes);
+        # the Confluent SR is attached to it in setUp. Mirrors the Kafka-source
+        # topology used by cluster_linking_e2e_test.
+        return SecondaryClusterSpec(
+            ServiceType.KAFKA, kafka_version="3.8.0", kafka_quorum="COMBINED_KRAFT"
+        )
+
+    def setUp(self):
+        # Starts the Apache Kafka source + Redpanda destination clusters.
+        super().setUp()
+        # Attach a Confluent Schema Registry to the Kafka source and start it.
+        self._confluent_sr = ConfluentSchemaRegistryService(
+            self.test_context, bootstrap_provider=self.source_cluster_service
+        )
+        self._confluent_sr.start()
+
+    def _confluent(self) -> ConfluentSchemaRegistryService:
+        assert self._confluent_sr is not None, "Confluent SR not started"
+        return self._confluent_sr
+
+    def _make_source_client(self) -> SchemaRegistryRedpandaClient:
+        sr = self._confluent()
+        # The base client hardcodes port 8081 and only needs
+        # .nodes[*].account.hostname + .logger, so the ducktape service
+        # duck-types cleanly; the Confluent SR must use the default port.
+        assert sr.port == 8081, "Confluent SR must listen on 8081 for the SR client"
+        return SchemaRegistryRedpandaClient(sr)  # type: ignore[arg-type]

tests/rptest/tests/cluster_linking_test_base.py

@@ -585,6 +585,11 @@ class ShadowLinkTestBase(PreallocNodesTest):
     the target cluster. Secondary service is used as the source cluster.
     """
 
+    # Broker count for the secondary (source) cluster. Subclasses can shrink
+    # this to stay within their @cluster(num_nodes=...) budget when the source
+    # topology needs extra nodes (e.g. a separate Schema Registry service).
+    secondary_num_brokers: int = 3
+
     def __init__(
         self,
         test_context: TestContext,
@@ -741,7 +746,7 @@ def setUp(self):
             self.logger,
             self.redpanda,
             secondary_spec=self.source_cluster_spec,
-            num_brokers=3,
+            num_brokers=self.secondary_num_brokers,
             secondary_args=self.secondary_cluster_args,
         )
         self.services.setUp()