|
| 1 | +# Copyright 2026 Redpanda Data, Inc. |
| 2 | +# |
| 3 | +# Use of this software is governed by the Business Source License |
| 4 | +# included in the file licenses/BSL.md |
| 5 | +# |
| 6 | +# As of the Change Date specified in that file, in accordance with |
| 7 | +# the Business Source License, use of this software will be governed |
| 8 | +# by the Apache License, Version 2.0 |
| 9 | + |
| 10 | +import os |
| 11 | +import tempfile |
| 12 | + |
| 13 | +import requests |
| 14 | +from ducktape.services.service import Service |
| 15 | +from ducktape.utils.util import wait_until |
| 16 | + |
| 17 | +CP_DIR = "/opt/confluent" |
| 18 | +SR_BIN = os.path.join(CP_DIR, "bin", "schema-registry-start") |
| 19 | +SR_LOG = "/var/log/schema-registry.log" |
| 20 | +SR_PROPS = "/tmp/schema-registry.properties" |
| 21 | +SR_DEFAULT_PORT = 8081 |
| 22 | + |
| 23 | + |
| 24 | +class ConfluentSchemaRegistryService(Service): |
| 25 | + """ |
| 26 | + Single-node Confluent Schema Registry, used for migration / cross-vendor |
| 27 | + compatibility tests. The registry stores schemas in the `_schemas` topic |
| 28 | + of whichever Kafka-compatible cluster is passed as `bootstrap_provider`. |
| 29 | +
|
| 30 | + `bootstrap_provider` is anything that exposes a `bootstrap_servers()` |
| 31 | + method returning a comma-separated host:port list — e.g. a |
| 32 | + `KafkaServiceAdapter` wrapping `kafkatest.services.kafka.KafkaService`, |
| 33 | + or a `RedpandaService`. |
| 34 | + """ |
| 35 | + |
| 36 | + logs = { |
| 37 | + "schema_registry_log": { |
| 38 | + "path": SR_LOG, |
| 39 | + "collect_default": True, |
| 40 | + }, |
| 41 | + } |
| 42 | + |
| 43 | + def __init__(self, context, bootstrap_provider, port: int = SR_DEFAULT_PORT): |
| 44 | + super().__init__(context, num_nodes=1) |
| 45 | + self._bootstrap_provider = bootstrap_provider |
| 46 | + self.port = port |
| 47 | + |
| 48 | + def _bootstrap_servers(self) -> str: |
| 49 | + bs = self._bootstrap_provider.bootstrap_servers() |
| 50 | + # Confluent SR expects PLAINTEXT://host:port form for plaintext. |
| 51 | + return ",".join( |
| 52 | + f"PLAINTEXT://{ep}" if "://" not in ep else ep for ep in bs.split(",") |
| 53 | + ) |
| 54 | + |
| 55 | + def _properties(self) -> str: |
| 56 | + return ( |
| 57 | + f"listeners=http://0.0.0.0:{self.port}\n" |
| 58 | + f"kafkastore.bootstrap.servers={self._bootstrap_servers()}\n" |
| 59 | + "kafkastore.topic=_schemas\n" |
| 60 | + "schema.compatibility.level=none\n" |
| 61 | + # This is a test-only registry; RF=1 keeps `_schemas` writable on a |
| 62 | + # small source cluster regardless of its broker count. |
| 63 | + "kafkastore.topic.replication.factor=1\n" |
| 64 | + ) |
| 65 | + |
| 66 | + def url(self, node=None) -> str: |
| 67 | + node = node or self.nodes[0] |
| 68 | + return f"http://{node.account.hostname}:{self.port}" |
| 69 | + |
| 70 | + def alive(self, node) -> bool: |
| 71 | + try: |
| 72 | + r = requests.get(f"{self.url(node)}/subjects", timeout=2) |
| 73 | + return r.status_code == 200 |
| 74 | + except Exception: |
| 75 | + return False |
| 76 | + |
| 77 | + def pids(self, node): |
| 78 | + # The registry's JVM main class is |
| 79 | + # io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain; match on |
| 80 | + # the (unambiguous) simple class name so stop_node actually finds it. |
| 81 | + return node.account.java_pids("SchemaRegistryMain") |
| 82 | + |
| 83 | + def start_node(self, node, **kwargs): |
| 84 | + props = self._properties() |
| 85 | + self.logger.info(f"Starting Confluent SR on {node.account.hostname}\n{props}") |
| 86 | + with tempfile.NamedTemporaryFile(mode="w") as f: |
| 87 | + f.write(props) |
| 88 | + f.flush() |
| 89 | + node.account.copy_to(f.name, SR_PROPS) |
| 90 | + |
| 91 | + cmd = f"nohup {SR_BIN} {SR_PROPS} > {SR_LOG} 2>&1 &" |
| 92 | + node.account.ssh(cmd, allow_fail=False) |
| 93 | + |
| 94 | + wait_until( |
| 95 | + lambda: self.alive(node), |
| 96 | + timeout_sec=90, |
| 97 | + backoff_sec=2, |
| 98 | + err_msg=f"Confluent SR did not become reachable at {self.url(node)}", |
| 99 | + ) |
| 100 | + self.logger.info(f"Confluent SR is up at {self.url(node)}") |
| 101 | + |
| 102 | + def stop_node(self, node, **kwargs): |
| 103 | + for pid in self.pids(node): |
| 104 | + node.account.signal(pid, 15, allow_fail=True) |
| 105 | + wait_until( |
| 106 | + lambda: len(self.pids(node)) == 0, |
| 107 | + timeout_sec=30, |
| 108 | + backoff_sec=1, |
| 109 | + err_msg="Confluent SR did not stop", |
| 110 | + ) |
| 111 | + |
| 112 | + def clean_node(self, node, **kwargs): |
| 113 | + node.account.ssh(f"rm -f {SR_PROPS} {SR_LOG}", allow_fail=True) |
0 commit comments