datalake: add e2e test for per-topic schema registry context

Add a ducktape integration test verifying the full end-to-end path for
the `redpanda.schema.registry.context` topic property: SR schema
registration in contexts, topic property configuration, translator
schema resolution, and typed Iceberg columns.

test_context_isolation: two topics bound to different SR contexts resolve
different schemas from the same numeric schema ID, producing different
Iceberg table column layouts.

test_wrong_context_dlq: schema ID not present in the configured context
sends records to the dead-letter-queue table.

Author: wdberkeley

tests/rptest/clients/types.py

@@ -49,6 +49,7 @@ class TopicSpec:
     PROPERTY_ICEBERG_INVALID_RECORD_ACTION = "redpanda.iceberg.invalid.record.action"
     PROPERTY_ICEBERG_TARGET_LAG_MS = "redpanda.iceberg.target.lag.ms"
     PROPERTY_ICEBERG_PARTITION_SPEC = "redpanda.iceberg.partition.spec"
+    PROPERTY_SCHEMA_REGISTRY_CONTEXT = "redpanda.schema.registry.context"
     PROPERTY_MIN_CLEANABLE_DIRTY_RATIO = "min.cleanable.dirty.ratio"
     PROPERTY_MIN_COMPACTION_LAG_MS = "min.compaction.lag.ms"
     PROPERTY_MAX_COMPACTION_LAG_MS = "max.compaction.lag.ms"

tests/rptest/tests/datalake/schema_registry_context_test.py

@@ -0,0 +1,237 @@
+# Copyright 2025 Redpanda Data, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+
+import io
+import json
+import struct
+
+import avro.io
+import avro.schema
+import confluent_kafka
+import requests
+from ducktape.mark import matrix
+
+from rptest.clients.types import TopicSpec
+from rptest.services.cluster import cluster
+from rptest.services.redpanda import SISettings, SchemaRegistryConfig
+from rptest.tests.datalake.catalog_service_factory import (
+    filesystem_catalog_type,
+)
+from rptest.tests.datalake.datalake_services import DatalakeServices
+from rptest.tests.datalake.query_engine_base import QueryEngineType
+from rptest.tests.datalake.utils import supported_storage_types
+from rptest.tests.redpanda_test import RedpandaTest
+
+SCHEMA_A = {
+    "type": "record",
+    "name": "RecordA",
+    "fields": [
+        {"name": "name", "type": "string"},
+        {"name": "age", "type": "int"},
+    ],
+}
+
+SCHEMA_B = {
+    "type": "record",
+    "name": "RecordB",
+    "fields": [
+        {"name": "color", "type": "string"},
+        {"name": "size", "type": "double"},
+    ],
+}
+
+SCHEMA_C = {
+    "type": "record",
+    "name": "RecordC",
+    "fields": [
+        {"name": "x", "type": "int"},
+    ],
+}
+
+
+class DatalakeSchemaRegistryContextTest(RedpandaTest):
+    def __init__(self, test_context):
+        super().__init__(
+            test_context=test_context,
+            num_brokers=1,
+            extra_rp_conf={
+                "iceberg_enabled": True,
+                "iceberg_catalog_commit_interval_ms": 5000,
+                "schema_registry_enable_qualified_subjects": True,
+            },
+            schema_registry_config=SchemaRegistryConfig(),
+            si_settings=SISettings(test_context=test_context),
+        )
+
+    def setUp(self):
+        # DatalakeServices starts Redpanda.
+        pass
+
+    def _sr_url(self):
+        return self.redpanda.schema_reg().split(",")[0]
+
+    def _register_schema(self, context, subject, schema_dict):
+        """Register an Avro schema in the given SR context. Returns the
+        assigned schema ID."""
+        url = f"{self._sr_url()}/subjects/:.{context}:{subject}/versions"
+        resp = requests.post(
+            url,
+            headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
+            json={"schema": json.dumps(schema_dict), "schemaType": "AVRO"},
+        )
+        resp.raise_for_status()
+        return resp.json()["id"]
+
+    def _make_confluent_record(self, schema_id, schema_dict, record):
+        """Build a Confluent wire-format payload: magic byte + 4-byte
+        schema ID + Avro binary-encoded record."""
+        parsed = avro.schema.parse(json.dumps(schema_dict))
+        buf = io.BytesIO()
+        buf.write(struct.pack(">bI", 0, schema_id))
+        encoder = avro.io.BinaryEncoder(buf)
+        writer = avro.io.DatumWriter(parsed)
+        writer.write(record, encoder)
+        return buf.getvalue()
+
+    def _produce_confluent_records(self, topic, schema_id, schema_dict, records):
+        """Produce raw Confluent wire-format records via confluent_kafka."""
+        producer = confluent_kafka.Producer(
+            {"bootstrap.servers": self.redpanda.brokers()}
+        )
+        for record in records:
+            payload = self._make_confluent_record(schema_id, schema_dict, record)
+            producer.produce(topic, value=payload)
+        producer.flush()
+
+    @cluster(num_nodes=3)
+    @matrix(cloud_storage_type=supported_storage_types())
+    def test_context_isolation(self, cloud_storage_type):
+        """Two topics bound to different SR contexts resolve different
+        schemas from the same numeric schema ID, producing different
+        Iceberg column layouts."""
+
+        with DatalakeServices(
+            self.test_context,
+            redpanda=self.redpanda,
+            catalog_type=filesystem_catalog_type(),
+            include_query_engines=[QueryEngineType.SPARK],
+        ) as dl:
+            # Register schemas in separate contexts.
+            id_a = self._register_schema("ctx1", "topic_a-value", SCHEMA_A)
+            id_b = self._register_schema("ctx2", "topic_b-value", SCHEMA_B)
+            self.logger.info(
+                f"Registered schema A (id={id_a}) in .ctx1, "
+                f"schema B (id={id_b}) in .ctx2"
+            )
+            # SR assigns IDs per-context starting at 1, so both should
+            # get the same numeric ID. This is what makes the test
+            # meaningful: it proves that the translator resolves the
+            # correct schema via (context, id) rather than just id.
+            # May become brittle if SR changes to global ID allocation.
+            assert id_a == id_b, (
+                f"Expected same schema ID in both contexts, "
+                f"got id_a={id_a}, id_b={id_b}"
+            )
+
+            # Create topics with per-topic SR context.
+            dl.create_iceberg_enabled_topic(
+                "topic_a",
+                iceberg_mode="value_schema_id_prefix",
+                config={
+                    TopicSpec.PROPERTY_SCHEMA_REGISTRY_CONTEXT: ".ctx1",
+                },
+            )
+            dl.create_iceberg_enabled_topic(
+                "topic_b",
+                iceberg_mode="value_schema_id_prefix",
+                config={
+                    TopicSpec.PROPERTY_SCHEMA_REGISTRY_CONTEXT: ".ctx2",
+                },
+            )
+
+            # Produce records.
+            records_a = [{"name": f"user_{i}", "age": 20 + i} for i in range(10)]
+            records_b = [{"color": f"color_{i}", "size": float(i)} for i in range(10)]
+
+            self._produce_confluent_records("topic_a", id_a, SCHEMA_A, records_a)
+            self._produce_confluent_records("topic_b", id_b, SCHEMA_B, records_b)
+
+            # Wait for translation.
+            dl.wait_for_translation("topic_a", msg_count=10)
+            dl.wait_for_translation("topic_b", msg_count=10)
+
+            # Verify Iceberg table columns.
+            spark = dl.spark()
+
+            desc_a = spark.run_query_fetch_all("describe redpanda.topic_a")
+            # Spark describe returns header row at [0] and partition info
+            # in the last 3 rows; strip those.
+            cols_a = {(r[0], r[1]) for r in desc_a[1:-3]}
+            assert ("name", "string") in cols_a, (
+                f"Expected 'name' string column in topic_a, got {cols_a}"
+            )
+            assert ("age", "int") in cols_a, (
+                f"Expected 'age' int column in topic_a, got {cols_a}"
+            )
+
+            desc_b = spark.run_query_fetch_all("describe redpanda.topic_b")
+            cols_b = {(r[0], r[1]) for r in desc_b[1:-3]}
+            assert ("color", "string") in cols_b, (
+                f"Expected 'color' string column in topic_b, got {cols_b}"
+            )
+            assert ("size", "double") in cols_b, (
+                f"Expected 'size' double column in topic_b, got {cols_b}"
+            )
+
+            # Negative: topic_a should NOT have topic_b's columns.
+            assert ("color", "string") not in cols_a, (
+                "topic_a should not have topic_b's 'color' column"
+            )
+            assert ("name", "string") not in cols_b, (
+                "topic_b should not have topic_a's 'name' column"
+            )
+
+    @cluster(num_nodes=3)
+    @matrix(cloud_storage_type=supported_storage_types())
+    def test_wrong_context_dlq(self, cloud_storage_type):
+        """Schema ID not present in the configured context sends records
+        to the dead-letter-queue table."""
+
+        with DatalakeServices(
+            self.test_context,
+            redpanda=self.redpanda,
+            catalog_type=filesystem_catalog_type(),
+            include_query_engines=[QueryEngineType.SPARK],
+        ) as dl:
+            # Register schema in .ctx1 only.
+            schema_id = self._register_schema("ctx1", "topic_c-value", SCHEMA_C)
+            self.logger.info(f"Registered schema C (id={schema_id}) in .ctx1")
+
+            # Create topic pointing to a context where the schema does
+            # not exist.
+            dl.create_iceberg_enabled_topic(
+                "topic_c",
+                iceberg_mode="value_schema_id_prefix",
+                config={
+                    TopicSpec.PROPERTY_SCHEMA_REGISTRY_CONTEXT: ".wrong",
+                    TopicSpec.PROPERTY_ICEBERG_INVALID_RECORD_ACTION: "dlq_table",
+                },
+            )
+
+            # Produce records — translator will fail to resolve schema_id
+            # in .wrong context.
+            records = [{"x": i} for i in range(10)]
+            self._produce_confluent_records("topic_c", schema_id, SCHEMA_C, records)
+
+            # Records should land in the DLQ table.
+            dl.wait_for_translation(
+                "topic_c",
+                msg_count=10,
+                table_override="topic_c~dlq",
+            )

tools/type-checking/type-check-strictness.json

@@ -128,6 +128,7 @@
         "rptest/tests/datalake/protobuf_json_test.py",
         "rptest/tests/datalake/recovery_mode_test.py",
         "rptest/tests/datalake/rest_catalog_connection_test.py",
+        "rptest/tests/datalake/schema_registry_context_test.py",
         "rptest/tests/datalake/simple_connect_test.py",
         "rptest/tests/datalake/spark_sql_server_smoke_test.py",
         "rptest/tests/datalake/throttling_test.py",