diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index 9883559..15a4a72 100644
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@@ -72,6 +72,7 @@ stackstate-backup-cli/
 - `cmd/victoriametrics/`: VictoriaMetrics backup/restore commands (list, restore, check-and-finalize)
 - `cmd/settings/`: Settings backup/restore commands (list, restore, check-and-finalize)
 - `cmd/version/`: Version information
+- `cmd/replication/`: Read-only HA replication observations and bounded waiting
 
 **Dependency Rules**:
 - ✅ Can import: `internal/app/*` (preferred), all other `internal/` packages
@@ -126,6 +127,7 @@ appCtx.NewCHClient(backupAPIPort, dbPort) // ClickHouse client factory
 
 **Key Packages**:
 - `portforward/`: Manages Kubernetes port-forwarding lifecycle
+- `replication/`: Discovers namespace workloads and evaluates fixed database queries, without backup configuration or restore operations
 - `scale/`: Deployment and StatefulSet scaling workflows with detailed logging
 - `restore/`: Restore job orchestration (confirmation, job lifecycle, finalization, resource management)
 - `restorelock/`: Prevents parallel restore operations using Kubernetes annotations
diff --git a/README.md b/README.md
index bac5f87..fb39ac6 100644
--- a/README.md
+++ b/README.md
@@ -12,6 +12,7 @@ This CLI tool replaces the legacy Bash-based backup/restore scripts with a singl
 - Stackgraph backups and restores
 - VictoriaMetrics backups and restores
 - Settings backups and restores
+- Read-only HA database replication checks
 
 ## Installation
 
@@ -44,6 +45,19 @@ sts-backup [command] [subcommand] [flags]
 
 ## Commands
 
+### replication check
+
+Inspect HDFS, Elasticsearch, Kafka, ClickHouse and ZooKeeper replication in one namespace:
+
+```bash
+sts-backup replication check --namespace observability
+sts-backup replication check --namespace observability --wait --output json
+```
+
+This command has its own flags and does not require backup configuration.
+See [Replication checks](docs/replication.md) for status and exit-code semantics,
+authentication, a Kubernetes Job example, and the maintenance checks outside its scope.
+
 ### version
 
 Display version information.
diff --git a/cmd/replication/replication.go b/cmd/replication/replication.go
new file mode 100644
index 0000000..bfcb207
--- /dev/null
+++ b/cmd/replication/replication.go
@@ -0,0 +1,186 @@
+// Package replication exposes read-only database replication checks.
+package replication
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"text/tabwriter"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/stackvista/stackstate-backup-cli/internal/app"
+	checker "github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication"
+)
+
+const (
+	defaultTimeout        = 10 * time.Minute
+	defaultRequestTimeout = 30 * time.Second
+	defaultInterval       = 10 * time.Second
+	defaultStableFor      = 30 * time.Second
+	tablePadding          = 2
+)
+
+type flags struct {
+	options    checker.Options
+	kubeconfig string
+	output     string
+	wait       bool
+	timeout    time.Duration
+	interval   time.Duration
+	stableFor  time.Duration
+}
+
+// Cmd creates the replication command independently of backup configuration.
+func Cmd() *cobra.Command {
+	command := &cobra.Command{Use: "replication", Short: "Inspect HA database replication without changing cluster state"}
+	f := &flags{}
+	check := &cobra.Command{
+		Use: "check", Short: "Check observed replication; return nonzero unless all selected checks pass",
+		Long: "Check chart-managed HDFS, Elasticsearch, Kafka, ClickHouse and ZooKeeper replication. " +
+			"This is a point-in-time observation, not permission to remove a node. " +
+			"Requires pods/exec access; only fixed read-only database queries are executed.",
+		Args: cobra.NoArgs, SilenceUsage: true,
+		RunE: func(command *cobra.Command, _ []string) error { return run(command, f) },
+	}
+	check.Flags().StringVarP(&f.options.Namespace, "namespace", "n", "", "Kubernetes namespace (required)")
+	check.Flags().StringVar(&f.kubeconfig, "kubeconfig", "", "Kubeconfig path; uses normal kubeconfig or in-cluster credentials")
+	check.Flags().StringSliceVar(&f.options.Components, "components", []string{"hdfs", "elasticsearch", "kafka", "clickhouse", "zookeeper"}, "Components to check")
+	check.Flags().StringVarP(&f.output, "output", "o", "table", "Output format: table or json")
+	check.Flags().BoolVar(&f.wait, "wait", false, "Wait for sustained healthy replication")
+	check.Flags().DurationVar(&f.timeout, "timeout", defaultTimeout, "Overall deadline, including queries")
+	check.Flags().DurationVar(&f.options.RequestTimeout, "request-timeout", defaultRequestTimeout, "Deadline for each Kubernetes request or database query")
+	check.Flags().DurationVar(&f.interval, "interval", defaultInterval, "Interval between observations in wait mode")
+	check.Flags().DurationVar(&f.stableFor, "stable-for", defaultStableFor, "Required healthy observation period in wait mode")
+	check.Flags().StringVar(&f.options.KafkaClientProperties, "kafka-client-properties", "", "Kafka client properties file already mounted in broker pods")
+	check.Flags().StringVar(&f.options.KafkaBootstrapServer, "kafka-bootstrap-server", "localhost:9092", "Kafka bootstrap address reachable from the broker pod")
+	check.Flags().StringVar(&f.options.ElasticsearchScheme, "elasticsearch-scheme", "http", "Elasticsearch loopback protocol: http or https")
+	check.Flags().StringVar(&f.options.ElasticsearchCA, "elasticsearch-ca", "", "CA file already mounted in Elasticsearch pods")
+	check.Flags().StringVar(&f.options.ElasticsearchHost, "elasticsearch-server-name", "127.0.0.1", "Elasticsearch TLS server name, resolved to loopback inside the pod")
+	_ = check.MarkFlagRequired("namespace")
+	command.AddCommand(check)
+	return command
+}
+
+func (f *flags) validate() error {
+	if f.output != "table" && f.output != "json" {
+		return fmt.Errorf("output must be table or json")
+	}
+	if f.timeout <= 0 || f.interval <= 0 || f.stableFor < 0 {
+		return fmt.Errorf("timeout and interval must be positive; stable-for cannot be negative")
+	}
+	if f.wait && f.stableFor >= f.timeout {
+		return fmt.Errorf("stable-for must be shorter than timeout")
+	}
+	return f.options.Validate()
+}
+
+func run(command *cobra.Command, f *flags) error {
+	if err := f.validate(); err != nil {
+		return err
+	}
+	probe, err := app.NewReplicationChecker(f.kubeconfig, f.options)
+	if err != nil {
+		return err
+	}
+	ctx, stop := signal.NotifyContext(command.Context(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	ctx, cancel := context.WithTimeout(ctx, f.timeout)
+	defer cancel()
+	report, checkErr := observe(ctx, probe.Check, f, command.ErrOrStderr())
+	return finishReport(command.OutOrStdout(), f.output, report, checkErr)
+}
+
+func finishReport(writer io.Writer, format string, report checker.Report, checkErr error) error {
+	if checkErr != nil {
+		report.Status = checker.Unknown
+		report.Error = checkErr.Error()
+	}
+	if err := writeReport(writer, format, report); err != nil {
+		return err
+	}
+	if checkErr != nil {
+		return checkErr
+	}
+	if report.Status != checker.Healthy {
+		return fmt.Errorf("replication is %s; see the report", report.Status)
+	}
+	return nil
+}
+
+func observe(ctx context.Context, check func(context.Context) checker.Report, f *flags, progress io.Writer) (checker.Report, error) {
+	if !f.wait {
+		report := check(ctx)
+		if ctx.Err() != nil {
+			return checker.Report{Namespace: report.Namespace}, fmt.Errorf("replication check ended: %w", ctx.Err())
+		}
+		return report, nil
+	}
+	_, _ = fmt.Fprintf(progress, "[%s] Waiting for all checks to remain healthy for %s (timeout %s).\n",
+		time.Now().UTC().Format(time.RFC3339), f.stableFor, f.timeout)
+	return checker.Wait(ctx, check, f.interval, f.stableFor, func(report checker.Report, state checker.WaitProgress) {
+		timestamp := state.ObservedAt.Format(time.RFC3339)
+		for _, check := range report.Checks {
+			_, _ = fmt.Fprintf(progress, "[%s] %s %s: %s\n", timestamp, check.Component, check.Status, strings.Join(check.Messages, "; "))
+		}
+		_, _ = fmt.Fprintf(progress, "[%s] %s\n", timestamp, stabilityMessage(report, state))
+	})
+}
+
+func stabilityMessage(report checker.Report, state checker.WaitProgress) string {
+	switch {
+	case state.Complete:
+		return fmt.Sprintf("All checks healthy; stability period satisfied (%s/%s).",
+			state.HealthyFor.Round(time.Millisecond), state.Required)
+	case report.Status == checker.Healthy:
+		return fmt.Sprintf("All checks healthy; verifying stability: %s/%s (%s remaining).",
+			state.HealthyFor.Round(time.Millisecond), state.Required, (state.Required - state.HealthyFor).Round(time.Millisecond))
+	case state.Reset:
+		return "Stability period reset; waiting for all selected checks to become healthy."
+	default:
+		return "Waiting for all selected checks to become healthy; stability period has not started."
+	}
+}
+
+func writeReport(writer io.Writer, format string, report checker.Report) error {
+	if format == "json" {
+		if err := json.NewEncoder(writer).Encode(report); err != nil {
+			return fmt.Errorf("write JSON report: %w", err)
+		}
+		return nil
+	}
+	if _, err := fmt.Fprintf(writer, "[%s] Replication result: %s\n", time.Now().UTC().Format(time.RFC3339), report.Status); err != nil {
+		return fmt.Errorf("write report status: %w", err)
+	}
+	if report.Error != "" {
+		if _, err := fmt.Fprintln(writer, report.Error); err != nil {
+			return fmt.Errorf("write report error: %w", err)
+		}
+	}
+	if len(report.Checks) == 0 {
+		_, err := fmt.Fprintln(writer, "No completed observation.")
+		return err
+	}
+	if _, err := fmt.Fprintf(writer, "Last completed observation started: %s\n", report.CheckedAt.Format(time.RFC3339)); err != nil {
+		return fmt.Errorf("write observation timestamp: %w", err)
+	}
+	table := tabwriter.NewWriter(writer, 0, 0, tablePadding, ' ', 0)
+	if _, err := fmt.Fprintln(table, "COMPONENT\tSTATUS\tDETAILS"); err != nil {
+		return fmt.Errorf("write report header: %w", err)
+	}
+	for _, check := range report.Checks {
+		if _, err := fmt.Fprintf(table, "%s\t%s\t%s\n", check.Component, check.Status, strings.Join(check.Messages, "; ")); err != nil {
+			return fmt.Errorf("write report row: %w", err)
+		}
+	}
+	if err := table.Flush(); err != nil {
+		return fmt.Errorf("flush report: %w", err)
+	}
+	return nil
+}
diff --git a/cmd/replication/replication_test.go b/cmd/replication/replication_test.go
new file mode 100644
index 0000000..c2fd8d7
--- /dev/null
+++ b/cmd/replication/replication_test.go
@@ -0,0 +1,127 @@
+package replication
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	checker "github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication"
+)
+
+func TestJSONOutputIsSeparateFromProgress(t *testing.T) {
+	var stdout, stderr bytes.Buffer
+	report, err := observe(context.Background(), func(context.Context) checker.Report {
+		return checker.Report{Status: checker.Healthy, Checks: []checker.Result{
+			{Component: "kafka", Status: checker.Healthy, Messages: []string{"all assigned replicas in sync"}},
+		}}
+	}, &flags{wait: true, interval: time.Millisecond}, &stderr)
+	require.NoError(t, err)
+	require.NoError(t, writeReport(&stdout, "json", report))
+	var decoded checker.Report
+	require.NoError(t, json.Unmarshal(stdout.Bytes(), &decoded))
+	assert.Equal(t, checker.Healthy, decoded.Status)
+	assert.Contains(t, stderr.String(), "kafka healthy")
+}
+
+func TestCheckValidatesBeforeConnecting(t *testing.T) {
+	command := Cmd()
+	command.SetArgs([]string{"check", "--namespace=test", "--components=kafka", "--output=invalid"})
+	command.SetOut(&bytes.Buffer{})
+	command.SetErr(&bytes.Buffer{})
+	err := command.Execute()
+	require.ErrorContains(t, err, "output must be table or json")
+}
+
+func TestHelpDoesNotRequireBackupConfiguration(t *testing.T) {
+	command := Cmd()
+	command.SetArgs([]string{"check", "--help"})
+	var output bytes.Buffer
+	command.SetOut(&output)
+	require.NoError(t, command.Execute())
+	assert.NotContains(t, output.String(), "--release")
+	assert.NotContains(t, output.String(), "--secret")
+	assert.NotContains(t, output.String(), "--configmap")
+	assert.Contains(t, output.String(), "hdfs,elasticsearch,kafka,clickhouse,zookeeper")
+}
+
+func TestReportAndExitAgree(t *testing.T) {
+	tests := []struct {
+		name, status, expected string
+		checkErr               error
+	}{
+		{"healthy", checker.Healthy, checker.Healthy, nil},
+		{"degraded", checker.Degraded, checker.Degraded, nil},
+		{"unknown", checker.Unknown, checker.Unknown, nil},
+		{"timeout after healthy sample", checker.Healthy, checker.Unknown, context.DeadlineExceeded},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var output bytes.Buffer
+			err := finishReport(&output, "json", checker.Report{Status: test.status}, test.checkErr)
+			var report checker.Report
+			require.NoError(t, json.Unmarshal(output.Bytes(), &report))
+			assert.Equal(t, test.expected, report.Status)
+			if test.expected == checker.Healthy {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+			if test.checkErr != nil {
+				require.ErrorIs(t, err, test.checkErr)
+				assert.NotEmpty(t, report.Error)
+			}
+		})
+	}
+}
+
+func TestWaitOutputShowsTimestampsAndStabilityProgress(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var stderr bytes.Buffer
+		report, err := observe(context.Background(), func(context.Context) checker.Report {
+			return checker.Report{CheckedAt: time.Now().UTC(), Status: checker.Healthy,
+				Checks: []checker.Result{{Component: "kafka", Status: checker.Healthy, Messages: []string{"all replicas in sync"}}}}
+		}, &flags{wait: true, interval: 10 * time.Second, stableFor: 30 * time.Second, timeout: time.Minute}, &stderr)
+		require.NoError(t, err)
+		assert.Equal(t, checker.Healthy, report.Status)
+		for _, line := range strings.Split(strings.TrimSpace(stderr.String()), "\n") {
+			assert.Regexp(t, `^\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\] `, line)
+		}
+		assert.Contains(t, stderr.String(), "verifying stability: 0s/30s (30s remaining)")
+		assert.Contains(t, stderr.String(), "verifying stability: 10s/30s (20s remaining)")
+		assert.Contains(t, stderr.String(), "stability period satisfied (30s/30s)")
+	})
+}
+
+func TestCancelledTableLabelsLastCompletedObservation(t *testing.T) {
+	var stdout bytes.Buffer
+	report := checker.Report{CheckedAt: time.Date(2026, time.September, 11, 12, 0, 0, 0, time.UTC), Status: checker.Healthy,
+		Checks: []checker.Result{{Component: "kafka", Status: checker.Healthy, Messages: []string{"all replicas in sync"}}}}
+	err := finishReport(&stdout, "table", report, context.Canceled)
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Contains(t, stdout.String(), "Replication result: unknown")
+	assert.Contains(t, stdout.String(), "context canceled")
+	assert.Contains(t, stdout.String(), "Last completed observation started: 2026-09-11T12:00:00Z")
+	assert.Contains(t, stdout.String(), "all replicas in sync")
+}
+
+func TestCancellationBeforeFirstObservation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	var stdout, stderr bytes.Buffer
+	report, err := observe(ctx, func(context.Context) checker.Report {
+		cancel()
+		return checker.Report{Namespace: "test", Checks: []checker.Result{{Messages: []string{"aborted request URL"}}}}
+	}, &flags{wait: true, interval: time.Second}, &stderr)
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Equal(t, "test", report.Namespace)
+	assert.Empty(t, report.Checks)
+	assert.NotContains(t, stderr.String(), "aborted")
+	require.ErrorIs(t, finishReport(&stdout, "table", report, err), context.Canceled)
+	assert.Contains(t, stdout.String(), "No completed observation.")
+}
diff --git a/cmd/root.go b/cmd/root.go
index f2d7686..4ae0b28 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -6,6 +6,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/stackvista/stackstate-backup-cli/cmd/clickhouse"
 	"github.com/stackvista/stackstate-backup-cli/cmd/elasticsearch"
+	"github.com/stackvista/stackstate-backup-cli/cmd/replication"
 	"github.com/stackvista/stackstate-backup-cli/cmd/settings"
 	"github.com/stackvista/stackstate-backup-cli/cmd/stackgraph"
 	"github.com/stackvista/stackstate-backup-cli/cmd/version"
@@ -56,12 +57,13 @@ func init() {
 
 	// Add commands that don't need backup config flags
 	rootCmd.AddCommand(version.Cmd())
+	rootCmd.AddCommand(replication.Cmd())
 }
 
 var rootCmd = &cobra.Command{
 	Use:   "sts-backup",
-	Short: "Backup and restore tool for SUSE Observability platform",
-	Long:  `A CLI tool for managing backups and restores for SUSE Observability platform running on Kubernetes.`,
+	Short: "Backup, restore and replication checks for SUSE Observability",
+	Long:  `A CLI tool for managing backups, restores and database replication checks for SUSE Observability running on Kubernetes.`,
 }
 
 func Execute() {
diff --git a/docs/replication.md b/docs/replication.md
new file mode 100644
index 0000000..80a98ca
--- /dev/null
+++ b/docs/replication.md
@@ -0,0 +1,206 @@
+# Check database replication
+
+`sts-backup replication check` inspects the chart-managed HA databases in a
+namespace containing one SUSE Observability installation. It works from a
+workstation or a Kubernetes Job and does not require the backup ConfigMap,
+backup Secret, or enabled backups.
+
+```bash
+sts-backup replication check --namespace observability
+```
+
+The command returns exit code **0** only when every selected component reports
+healthy replication. Any degraded, missing, inaccessible, unsupported or
+incompletely described component returns exit code **1**. A missing component
+is never silently skipped. Non-HA installations do not meet the checker's
+minimum application replication requirement.
+
+Use JSON for automation:
+
+```bash
+sts-backup replication check \
+  --namespace observability \
+  --output json
+```
+
+The report identifies the namespace, observation time, selected
+components, status (`healthy`, `degraded` or `unknown`) and diagnostic messages.
+The status describes only the selected checks.
+
+## Wait for recovery
+
+```bash
+sts-backup replication check \
+  --namespace observability \
+  --wait --timeout 15m --interval 10s --stable-for 30s \
+  --output json > replication.json
+```
+
+Wait mode requires consecutive healthy observations spanning `--stable-for`.
+The default is 30 seconds, starting when the first fully healthy observation
+completes. Earlier rounds with any `unknown` or `degraded` component do not
+count. An unsuccessful observation resets the period.
+
+Progress includes UTC timestamps and shows the elapsed and remaining healthy
+period, resets, and successful completion. Seeing every component healthy
+does not mean the stability period has already elapsed. Database queries take
+time; the checker needs another completed healthy observation to confirm the
+period, rather than exiting on a timer alone.
+
+To finish on the first fully healthy observation, explicitly use
+`--wait --stable-for 0s`.
+
+Progress goes to stderr; stdout contains one final report. Table output includes
+the report time and the start time of the last completed observation. JSON
+retains its `checkedAt` timestamp. The overall deadline includes database
+queries. `--request-timeout` bounds each API request or query.
+
+Ctrl+C stops further queries. Cancellation or timeout returns nonzero and sets
+the overall result to `unknown`, retaining the last completed observation
+instead of replacing it with errors from interrupted queries. If no observation
+completed, the report says so. Healthy component results in a cancelled report
+describe that previous observation; the wait did not finish successfully.
+
+Select an explicit subset if a database is intentionally disabled:
+
+```bash
+sts-backup replication check \
+  --namespace observability \
+  --components hdfs,elasticsearch,kafka
+```
+
+This selection does not validate the omitted database.
+
+## What is checked
+
+The checker discovers StatefulSets using `app.kubernetes.io/name`
+(`hbase`, `elasticsearch`, `kafka`, `clickhouse`, `zookeeper`) and
+`app.kubernetes.io/component` within the namespace. The HDFS components
+`hdfs-nn` and `hdfs-dn` distinguish the NameNode and DataNodes from the
+SecondaryNameNode. It assumes one SUSE Observability installation per
+namespace; no Helm release name is required. It verifies the desired pods
+exist, belong to those StatefulSets, are Ready, and are not terminating or
+undergoing a rollout.
+It queries database state and invalidates the observation if Kubernetes
+resource versions change during those queries.
+
+| Component | Replication evidence |
+|---|---|
+| HDFS | Configured default block replication is at least two; all expected DataNodes are live; the NameNode is out of safe mode; no missing, corrupt, under-replicated or pending-replication blocks are reported by JMX. |
+| Elasticsearch | Expected members are present; health is green; every returned index has at least one replica shard; no shards are unassigned, initializing or relocating. |
+| Kafka | Every described partition has at least two distinct assigned replicas, complete ISR membership and an in-sync leader. Summaries and partition descriptions must agree. Both `__consumer_offsets` and `__transaction_state` must exist. |
+| ClickHouse | Every discovered member is queried. Replicated table groups have their expected active replicas, live coordination sessions and no read-only members. Replication logs are caught up, and no replication queue tasks other than background `MERGE_PARTS` remain. Missing or duplicate table replicas and current query exceptions fail the check. |
+| ZooKeeper | At least three voting members are available. Every member reports the expected voting membership; exactly one is leader and the rest are followers. The leader reports all expected followers synchronized and is checked again after sampling the ensemble. |
+
+The Kafka check does not create missing internal topics: initialize the
+corresponding workloads and repeat the check. The ClickHouse check requires
+replicated-table evidence and does not classify an empty result as healthy.
+Unreplicated ClickHouse tables are outside its scope.
+
+ClickHouse can retain `last_queue_update_exception` after recovery. The checker
+reports that history without failing otherwise healthy replication. Current
+coordination-query errors, expired sessions and replication backlog still fail.
+
+ZooKeeper is checked by default. To inspect it alone:
+
+```bash
+sts-backup replication check -n observability --components zookeeper --wait
+```
+
+ZooKeeper's `ruok` response does not prove that the ensemble has recovered.
+The checker reads `mntr` from every member instead, and requires full voting
+membership to recover rather than accepting a surviving majority. Wait mode
+applies the same healthy observation period as for the other databases.
+
+## Access and supported layouts
+
+The command uses the normal kubeconfig selection or a Kubernetes service
+account when running in a Pod. It requires listing Pods and StatefulSets and
+creating `pods/exec` requests in the target namespace.
+
+**`pods/exec` permission allows arbitrary commands in pods.** The checker
+itself executes only fixed read-only queries. Use a dedicated identity and
+grant this permission only in the installation's namespace. It does not scale
+workloads, change PDBs, annotate resources, repair databases or invoke restore
+operations. No backup credentials are loaded.
+
+Queries use the database tools already present in the product containers:
+`hdfs` and `curl` in the NameNode, `curl` in Elasticsearch,
+`kafka-topics.sh` in Kafka, `clickhouse-client` in ClickHouse, and Bash TCP
+access to ZooKeeper.
+Custom images, container names, external databases, alternative NameNode
+topologies and overridden database name/component labels are not supported by
+these initial adapters. Failed discovery or unsupported response formats
+produce `unknown`, not success.
+
+The initial HTTP adapters use the chart's NameNode and Elasticsearch ports.
+The ZooKeeper adapter requires the chart's plaintext loopback client port
+and `mntr` in its four-letter-command whitelist. It supports the chart's
+single ensemble of voting participants; external ensembles, observer layouts,
+weighted quorums and TLS-only client listeners are outside its scope. Missing
+membership or synchronization metrics produce `unknown`. No configuration
+changes, HTTP AdminServer or database writes are needed.
+
+For authenticated Kafka, supply a client properties file already mounted in
+the broker and an appropriate bootstrap address:
+
+```bash
+sts-backup replication check -n observability \
+  --components kafka \
+  --kafka-bootstrap-server suse-observability-kafka:9092 \
+  --kafka-client-properties /mounted/client.properties
+```
+
+The credentials must be able to describe all topics in the installation.
+Elasticsearch uses the pod's `ELASTIC_PASSWORD` when present. For HTTPS,
+use `--elasticsearch-scheme https`, `--elasticsearch-ca` with a CA path inside
+the pod, and `--elasticsearch-server-name` matching the server certificate.
+The server name is resolved to loopback inside that pod; certificate
+verification is not disabled.
+
+ClickHouse uses the pod's `CLICKHOUSE_ADMIN_USER`,
+`CLICKHOUSE_ADMIN_PASSWORD` and `CLICKHOUSE_TCP_PORT`. Credentials stay inside
+the pod. Query stderr is suppressed because database tools can echo credentials;
+when a query fails, inspect the component's configuration through your normal
+administrative procedure. Query output is size-limited and excess output is
+treated as unverified.
+
+## Kubernetes Job
+
+[examples/replication/job.yaml](../examples/replication/job.yaml) contains a
+dedicated ServiceAccount, namespace-scoped RBAC and a Job using in-cluster
+credentials. Adapt its namespace and image reference before use.
+The Job has no retries: a failure requires investigation and an explicit rerun.
+
+No new container image is published by this change. To package the CLI, build
+a static Linux binary from this repository and use the example SUSE BCI image:
+
+```bash
+CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o examples/replication/sts-backup .
+docker build -t registry.example.com/observability/sts-backup:replication-checker \
+  examples/replication
+```
+
+Publish the image to your registry through your normal image delivery process
+and replace the Job's example image reference. Use the architecture required
+by your nodes. The build copies the local binary; it does not download an
+unverified executable.
+
+## Maintenance boundary
+
+This is a sampled replication report, **not a “safe to remove this node”
+certificate or a maintenance lock**. It does not prevent another process from
+starting maintenance, provide an atomic database snapshot, or prove continued
+health between observations.
+
+This first version does not validate Longhorn volume health or placement,
+spare capacity, HBase region assignment and WAL recovery, quorum survival
+after a specific node is removed, VictoriaMetrics redundancy, backup freshness,
+or the consequences of removing a particular node. HDFS's default replication setting does not prove that
+every file has the same replication policy. It is not a complete implementation
+of the product's node-maintenance checklist.
+
+Continue to serialize maintenance, preserve storage redundancy, follow the
+documented recovery procedure and check these additional requirements.
+Run this checker after the affected node or replacement can schedule workloads.
+Only proceed when the report and the remaining maintenance checks pass.
diff --git a/examples/replication/.dockerignore b/examples/replication/.dockerignore
new file mode 100644
index 0000000..b80331d
--- /dev/null
+++ b/examples/replication/.dockerignore
@@ -0,0 +1,2 @@
+*
+!sts-backup
diff --git a/examples/replication/Dockerfile b/examples/replication/Dockerfile
new file mode 100644
index 0000000..0a1d5a2
--- /dev/null
+++ b/examples/replication/Dockerfile
@@ -0,0 +1,4 @@
+FROM registry.suse.com/bci/bci-micro:15.7@sha256:9e01097b36048042e276dd40e7661941ac4ba909237904bae99540eb90c9a5c6
+COPY sts-backup /usr/local/bin/sts-backup
+USER 65532:65532
+ENTRYPOINT ["/usr/local/bin/sts-backup"]
diff --git a/examples/replication/job.yaml b/examples/replication/job.yaml
new file mode 100644
index 0000000..34997df
--- /dev/null
+++ b/examples/replication/job.yaml
@@ -0,0 +1,76 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: replication-checker
+  namespace: observability
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: replication-checker
+  namespace: observability
+rules:
+  - apiGroups: [""]
+    resources: ["pods"]
+    verbs: ["get", "list"]
+  - apiGroups: ["apps"]
+    resources: ["statefulsets"]
+    verbs: ["get", "list"]
+  - apiGroups: [""]
+    resources: ["pods/exec"]
+    verbs: ["create"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: replication-checker
+  namespace: observability
+subjects:
+  - kind: ServiceAccount
+    name: replication-checker
+    namespace: observability
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: replication-checker
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: suse-observability-replication-check
+  namespace: observability
+spec:
+  backoffLimit: 0
+  activeDeadlineSeconds: 960
+  ttlSecondsAfterFinished: 86400
+  template:
+    spec:
+      serviceAccountName: replication-checker
+      restartPolicy: Never
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 65532
+        runAsGroup: 65532
+        seccompProfile:
+          type: RuntimeDefault
+      containers:
+        - name: checker
+          image: registry.example.com/observability/sts-backup:replication-checker
+          args:
+            - replication
+            - check
+            - --namespace=observability
+            - --wait
+            - --timeout=15m
+            - --stable-for=30s
+          resources:
+            requests:
+              cpu: 100m
+              memory: 128Mi
+            limits:
+              memory: 256Mi
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            capabilities:
+              drop: ["ALL"]
diff --git a/internal/app/replication.go b/internal/app/replication.go
new file mode 100644
index 0000000..0cf8937
--- /dev/null
+++ b/internal/app/replication.go
@@ -0,0 +1,17 @@
+package app
+
+import (
+	"fmt"
+
+	"github.com/stackvista/stackstate-backup-cli/internal/clients/k8s"
+	"github.com/stackvista/stackstate-backup-cli/internal/orchestration/replication"
+)
+
+// NewReplicationChecker requires Kubernetes access, but no backup ConfigMap or Secret.
+func NewReplicationChecker(kubeconfig string, options replication.Options) (*replication.Checker, error) {
+	client, err := k8s.NewClient(kubeconfig, false)
+	if err != nil {
+		return nil, fmt.Errorf("create Kubernetes client: %w", err)
+	}
+	return replication.New(client, options)
+}
diff --git a/internal/clients/k8s/exec.go b/internal/clients/k8s/exec.go
new file mode 100644
index 0000000..25336a3
--- /dev/null
+++ b/internal/clients/k8s/exec.go
@@ -0,0 +1,53 @@
+package k8s
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+const maxExecOutput = 8 << 20
+
+type boundedBuffer struct {
+	buffer bytes.Buffer
+	err    error
+}
+
+func (b *boundedBuffer) Write(p []byte) (int, error) {
+	if len(p) > maxExecOutput-b.buffer.Len() {
+		b.err = fmt.Errorf("query output exceeds %d bytes", maxExecOutput)
+		return 0, b.err
+	}
+	return b.buffer.Write(p)
+}
+
+// Exec runs a command without a TTY, bounded by the caller's context and output limit.
+func (c *Client) Exec(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) {
+	request := c.clientset.CoreV1().RESTClient().Post().
+		Namespace(namespace).Resource("pods").Name(pod).SubResource("exec").
+		VersionedParams(&corev1.PodExecOptions{
+			Container: container,
+			Command:   command,
+			Stdout:    true,
+			Stderr:    true,
+		}, scheme.ParameterCodec)
+	executor, err := remotecommand.NewSPDYExecutor(c.restConfig, http.MethodPost, request.URL())
+	if err != nil {
+		return nil, fmt.Errorf("create pod executor: %w", err)
+	}
+	var output boundedBuffer
+	// Database tools can echo authentication details in stderr.
+	if err := executor.StreamWithContext(ctx, remotecommand.StreamOptions{Stdout: &output, Stderr: io.Discard}); err != nil {
+		return nil, fmt.Errorf("query pod %s/%s: %w", namespace, pod, err)
+	}
+	if output.err != nil {
+		return nil, output.err
+	}
+	return output.buffer.Bytes(), nil
+}
diff --git a/internal/clients/k8s/exec_test.go b/internal/clients/k8s/exec_test.go
new file mode 100644
index 0000000..8f1cad2
--- /dev/null
+++ b/internal/clients/k8s/exec_test.go
@@ -0,0 +1,20 @@
+package k8s
+
+import (
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExecOutputLimit(t *testing.T) {
+	var output boundedBuffer
+	n, err := output.Write([]byte("report"))
+	require.NoError(t, err)
+	assert.Equal(t, 6, n)
+	_, err = io.Copy(&output, io.LimitReader(strings.NewReader(strings.Repeat("x", maxExecOutput)), int64(maxExecOutput)))
+	require.ErrorContains(t, err, "query output exceeds")
+	assert.LessOrEqual(t, output.buffer.Len(), maxExecOutput)
+}
diff --git a/internal/orchestration/replication/checker.go b/internal/orchestration/replication/checker.go
new file mode 100644
index 0000000..936cd1b
--- /dev/null
+++ b/internal/orchestration/replication/checker.go
@@ -0,0 +1,103 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// Checker observes databases without invoking backup or restore operations.
+type Checker struct {
+	kube    Kubernetes
+	options Options
+}
+
+// New creates a checker for the installation in one namespace.
+func New(kube Kubernetes, options Options) (*Checker, error) {
+	if err := options.Validate(); err != nil {
+		return nil, err
+	}
+	if options.KafkaBootstrapServer == "" {
+		options.KafkaBootstrapServer = "localhost:9092"
+	}
+	if options.ElasticsearchHost == "" {
+		options.ElasticsearchHost = "127.0.0.1"
+	}
+	return &Checker{kube: kube, options: options}, nil
+}
+
+// Check queries every selected component and rechecks Kubernetes membership afterward.
+func (c *Checker) Check(ctx context.Context) Report {
+	report := Report{
+		CheckedAt: time.Now().UTC(), Namespace: c.options.Namespace,
+		Checks: make([]Result, 0, len(c.options.Components)),
+	}
+	before, err := c.discover(ctx)
+	for _, component := range c.options.Components {
+		if ctx.Err() != nil {
+			report.Status = Unknown
+			return report
+		}
+		if err != nil {
+			report.Checks = append(report.Checks, result(component, Unknown, err.Error()))
+			continue
+		}
+		report.Checks = append(report.Checks, c.checkComponent(ctx, before, component))
+	}
+	if ctx.Err() != nil {
+		report.Status = Unknown
+		return report
+	}
+	if err == nil {
+		after, afterErr := c.discover(ctx)
+		if afterErr != nil || before.fingerprint() != after.fingerprint() {
+			message := "Kubernetes membership or status changed during the checks; repeat the observation"
+			if afterErr != nil {
+				message = afterErr.Error()
+			}
+			for n := range report.Checks {
+				report.Checks[n].Status = Unknown
+				report.Checks[n].Messages = append(report.Checks[n].Messages, message)
+			}
+		}
+	}
+	report.Status = reportStatus(report.Checks)
+	return report
+}
+
+func (c *Checker) query(ctx context.Context, pod, container string, command []string) ([]byte, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout)
+	defer cancel()
+	data, err := c.kube.Exec(ctx, c.options.Namespace, pod, container, command)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+	return data, err
+}
+
+func (c *Checker) checkComponent(ctx context.Context, inventory inventory, component string) Result {
+	switch component {
+	case "hdfs":
+		return c.checkHDFS(ctx, inventory)
+	case "elasticsearch":
+		return c.checkElasticsearch(ctx, inventory)
+	case "kafka":
+		return c.checkKafka(ctx, inventory)
+	case "clickhouse":
+		return c.checkClickHouse(ctx, inventory)
+	case "zookeeper":
+		return c.checkZooKeeper(ctx, inventory)
+	default:
+		return result(component, Unknown, "unsupported component")
+	}
+}
+
+func expectedMembers(members []member) error {
+	if len(members) < minReplicas {
+		return fmt.Errorf("at least %d application replicas are required; discovered %d", minReplicas, len(members))
+	}
+	return nil
+}
diff --git a/internal/orchestration/replication/checker_test.go b/internal/orchestration/replication/checker_test.go
new file mode 100644
index 0000000..b959c36
--- /dev/null
+++ b/internal/orchestration/replication/checker_test.go
@@ -0,0 +1,226 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/ptr"
+)
+
+type fakeKubernetes struct {
+	client kubernetes.Interface
+	exec   func(context.Context, string, string, string, []string) ([]byte, error)
+	calls  int
+}
+
+func (f *fakeKubernetes) Clientset() kubernetes.Interface { return f.client }
+
+func (f *fakeKubernetes) Exec(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) {
+	f.calls++
+	return f.exec(ctx, namespace, pod, container, command)
+}
+
+func testOptions() Options {
+	return Options{Namespace: "test", Components: []string{"kafka"}, RequestTimeout: time.Second, ElasticsearchScheme: "http"}
+}
+
+func kafkaObjects() []runtime.Object {
+	return databaseObjects("kafka", "kafka", "kafka", 2)
+}
+
+func databaseObjects(application, component, container string, replicas int32) []runtime.Object {
+	labels := map[string]string{
+		"app.kubernetes.io/name": application, "app.kubernetes.io/component": component, "app.kubernetes.io/instance": "arbitrary-release",
+	}
+	workload := &appsv1.StatefulSet{
+		ObjectMeta: metav1.ObjectMeta{Name: component, Namespace: "test", UID: types.UID("sts-" + component), Generation: 1, ResourceVersion: "1", Labels: labels},
+		Spec: appsv1.StatefulSetSpec{
+			Replicas: ptr.To(replicas),
+			Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: container}}}},
+		},
+		Status: appsv1.StatefulSetStatus{ReadyReplicas: replicas, ObservedGeneration: 1, CurrentRevision: "one", UpdateRevision: "one"},
+	}
+	objects := []runtime.Object{workload}
+	for n := int32(0); n < replicas; n++ {
+		name := fmt.Sprintf("%s-%d", component, n)
+		pod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: name, Namespace: "test", UID: types.UID(name), ResourceVersion: "1", Labels: labels,
+				OwnerReferences: []metav1.OwnerReference{{Kind: "StatefulSet", Name: component, UID: workload.UID, Controller: ptr.To(true)}},
+			},
+			Spec: corev1.PodSpec{NodeName: fmt.Sprintf("node-%d", n), Containers: []corev1.Container{{Name: container}}},
+			Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{
+				{Type: corev1.PodReady, Status: corev1.ConditionTrue},
+			}},
+		}
+		objects = append(objects, pod)
+	}
+	return objects
+}
+
+func TestCheckerQueriesReadyPodsAndDoesNotMutateKubernetes(t *testing.T) {
+	client := fake.NewSimpleClientset(kafkaObjects()...)
+	kube := &fakeKubernetes{client: client, exec: func(ctx context.Context, namespace, pod, container string, command []string) ([]byte, error) {
+		assert.Equal(t, "test", namespace)
+		assert.Equal(t, "kafka-0", pod)
+		assert.Equal(t, "kafka", container)
+		assert.Equal(t, []string{"bash", "-ec", kafkaQuery, "replication-check", "--bootstrap-server", "localhost:9092", "--describe"}, command)
+		_, deadline := ctx.Deadline()
+		assert.True(t, deadline)
+		return []byte(kafkaFixture()), nil
+	}}
+	probe, err := New(kube, testOptions())
+	require.NoError(t, err)
+	report := probe.Check(context.Background())
+	require.Equal(t, Healthy, report.Status, report)
+	assert.Equal(t, 1, kube.calls)
+	for _, action := range client.Actions() {
+		assert.Equal(t, "list", action.GetVerb())
+	}
+}
+
+func TestCheckerRejectsIncompleteTopology(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func([]runtime.Object) []runtime.Object
+	}{
+		{"missing pod", func(objects []runtime.Object) []runtime.Object { return objects[:2] }},
+		{"not ready", func(objects []runtime.Object) []runtime.Object {
+			objects[1].(*corev1.Pod).Status.Conditions[0].Status = corev1.ConditionFalse
+			return objects
+		}},
+		{"terminating", func(objects []runtime.Object) []runtime.Object {
+			objects[1].(*corev1.Pod).DeletionTimestamp = ptr.To(metav1.Now())
+			return objects
+		}},
+		{"wrong owner", func(objects []runtime.Object) []runtime.Object {
+			objects[1].(*corev1.Pod).OwnerReferences[0].UID = "other"
+			return objects
+		}},
+		{"wrong database label", func(objects []runtime.Object) []runtime.Object {
+			objects[0].(*appsv1.StatefulSet).Labels = map[string]string{"app.kubernetes.io/name": "other", "app.kubernetes.io/component": "kafka"}
+			return objects
+		}},
+		{"wrong component label", func(objects []runtime.Object) []runtime.Object {
+			objects[0].(*appsv1.StatefulSet).Labels["app.kubernetes.io/component"] = "unrelated"
+			return objects
+		}},
+		{"different namespace", func(objects []runtime.Object) []runtime.Object {
+			for _, object := range objects {
+				object.(metav1.Object).SetNamespace("other")
+			}
+			return objects
+		}},
+		{"rolling update", func(objects []runtime.Object) []runtime.Object {
+			objects[0].(*appsv1.StatefulSet).Status.UpdateRevision = "two"
+			return objects
+		}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			kube := &fakeKubernetes{client: fake.NewSimpleClientset(test.mutate(kafkaObjects())...)}
+			probe, err := New(kube, testOptions())
+			require.NoError(t, err)
+			assert.Equal(t, Unknown, probe.Check(context.Background()).Status)
+			assert.Zero(t, kube.calls)
+		})
+	}
+}
+
+func TestHDFSDiscoveryDistinguishesSecondaryNameNode(t *testing.T) {
+	objects := databaseObjects("hbase", "hdfs-nn", "namenode", 1)
+	objects = append(objects, databaseObjects("hbase", "hdfs-snn", "namenode", 1)...)
+	objects = append(objects, databaseObjects("hbase", "hdfs-dn", "datanode", 3)...)
+	kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects...), exec: func(_ context.Context, namespace, pod, container string, _ []string) ([]byte, error) {
+		assert.Equal(t, "test", namespace)
+		assert.Equal(t, "hdfs-nn-0", pod)
+		assert.Equal(t, "namenode", container)
+		return hdfsFixture(t, func(map[string]any) {}), nil
+	}}
+	options := testOptions()
+	options.Components = []string{"hdfs"}
+	probe, err := New(kube, options)
+	require.NoError(t, err)
+	report := probe.Check(context.Background())
+	assert.Equal(t, Healthy, report.Status, report)
+	assert.Equal(t, 1, kube.calls)
+}
+
+func TestDiscoveryDoesNotRequireHelmReleaseLabel(t *testing.T) {
+	objects := kafkaObjects()
+	for _, object := range objects {
+		delete(object.(metav1.Object).GetLabels(), "app.kubernetes.io/instance")
+	}
+	kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects...), exec: func(context.Context, string, string, string, []string) ([]byte, error) {
+		return []byte(kafkaFixture()), nil
+	}}
+	probe, err := New(kube, testOptions())
+	require.NoError(t, err)
+	assert.Equal(t, Healthy, probe.Check(context.Background()).Status)
+}
+
+func TestCheckerRejectsMembershipChangeDuringQueries(t *testing.T) {
+	client := fake.NewSimpleClientset(kafkaObjects()...)
+	kube := &fakeKubernetes{client: client, exec: func(ctx context.Context, namespace, pod, _ string, _ []string) ([]byte, error) {
+		current, err := client.CoreV1().Pods(namespace).Get(ctx, pod, metav1.GetOptions{})
+		require.NoError(t, err)
+		current.ResourceVersion = "2"
+		_, err = client.CoreV1().Pods(namespace).Update(ctx, current, metav1.UpdateOptions{})
+		require.NoError(t, err)
+		return []byte(kafkaFixture()), nil
+	}}
+	probe, err := New(kube, testOptions())
+	require.NoError(t, err)
+	report := probe.Check(context.Background())
+	assert.Equal(t, Unknown, report.Status)
+	assert.Contains(t, report.Checks[0].Messages, "Kubernetes membership or status changed during the checks; repeat the observation")
+}
+
+func TestQueryFailureCannotPass(t *testing.T) {
+	kube := &fakeKubernetes{client: fake.NewSimpleClientset(kafkaObjects()...), exec: func(context.Context, string, string, string, []string) ([]byte, error) {
+		return nil, fmt.Errorf("query not authorized")
+	}}
+	probe, err := New(kube, testOptions())
+	require.NoError(t, err)
+	assert.Equal(t, Unknown, probe.Check(context.Background()).Status)
+}
+
+func TestInvalidScopeRejected(t *testing.T) {
+	for _, components := range [][]string{nil, {"kafka", "kafka"}, {"not-a-store"}} {
+		options := testOptions()
+		options.Components = components
+		_, err := New(nil, options)
+		require.Error(t, err)
+	}
+}
+
+func TestCancellationStopsRemainingQueriesAndDiscovery(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	client := fake.NewSimpleClientset(kafkaObjects()...)
+	kube := &fakeKubernetes{client: client, exec: func(context.Context, string, string, string, []string) ([]byte, error) {
+		cancel()
+		return nil, fmt.Errorf("aborted request with query URL")
+	}}
+	options := testOptions()
+	options.Components = []string{"kafka", "zookeeper", "clickhouse"}
+	probe, err := New(kube, options)
+	require.NoError(t, err)
+	report := probe.Check(ctx)
+	assert.Equal(t, Unknown, report.Status)
+	require.Len(t, report.Checks, 1)
+	assert.Equal(t, []string{"context canceled"}, report.Checks[0].Messages)
+	assert.Equal(t, 1, kube.calls)
+	assert.Len(t, client.Actions(), 2, "no second discovery after cancellation")
+}
diff --git a/internal/orchestration/replication/clickhouse.go b/internal/orchestration/replication/clickhouse.go
new file mode 100644
index 0000000..6783f5e
--- /dev/null
+++ b/internal/orchestration/replication/clickhouse.go
@@ -0,0 +1,150 @@
+package replication
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+)
+
+const clickhouseSQL = `SELECT database, table, zookeeper_path, replica_name,
+is_readonly, is_session_expired, total_replicas, active_replicas,
+log_max_index, log_pointer, absolute_delay, last_queue_update_exception, zookeeper_exception,
+ifNull(pending_data_tasks, 0) AS pending_data_tasks
+FROM system.replicas
+LEFT JOIN (
+ SELECT database, table, countIf(type != 'MERGE_PARTS') AS pending_data_tasks
+ FROM system.replication_queue GROUP BY database, table
+) AS queues USING (database, table)
+WHERE database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema')
+FORMAT JSON`
+
+const clickhouseQuery = `export CLICKHOUSE_PASSWORD="${CLICKHOUSE_ADMIN_PASSWORD:?missing ClickHouse credentials}"
+exec clickhouse-client --host 127.0.0.1 --port "${CLICKHOUSE_TCP_PORT:-9000}" \
+ --user "${CLICKHOUSE_ADMIN_USER:?missing ClickHouse user}" --readonly 1 --query "$1"`
+
+type replicaObservation struct {
+	path               string
+	name               string
+	pod                string
+	total              int64
+	historicalQueueErr bool
+	problems           []string
+}
+
+func (c *Checker) checkClickHouse(ctx context.Context, inventory inventory) Result {
+	members, err := inventory.members("clickhouse", "clickhouse", "clickhouse")
+	if err != nil {
+		return result("clickhouse", Unknown, err.Error())
+	}
+	var observations []replicaObservation
+	for _, member := range members {
+		if member.replicas < minReplicas {
+			return result("clickhouse", Degraded, fmt.Sprintf("%s belongs to a shard with fewer than two replicas", member.pod.Name))
+		}
+		command := []string{"bash", "-ec", clickhouseQuery, "replication-check", clickhouseSQL}
+		data, err := c.query(ctx, member.pod.Name, "clickhouse", command)
+		if err != nil {
+			return result("clickhouse", Unknown, err.Error())
+		}
+		rows, err := parseClickHouse(data, member.pod.Name, member.replicas)
+		if err != nil {
+			return result("clickhouse", Unknown, fmt.Sprintf("%s: %v", member.pod.Name, err))
+		}
+		observations = append(observations, rows...)
+	}
+	return evaluateClickHouse(observations)
+}
+
+func parseClickHouse(data []byte, pod string, expected int) ([]replicaObservation, error) {
+	var response struct {
+		Data []map[string]json.RawMessage `json:"data"`
+	}
+	if err := json.Unmarshal(data, &response); err != nil || len(response.Data) == 0 {
+		return nil, fmt.Errorf("no replicated-table evidence returned")
+	}
+	var observations []replicaObservation
+	for _, row := range response.Data {
+		observation, err := parseReplica(row, pod, expected)
+		if err != nil {
+			return nil, err
+		}
+		observations = append(observations, observation)
+	}
+	return observations, nil
+}
+
+func parseReplica(row map[string]json.RawMessage, pod string, expected int) (replicaObservation, error) {
+	fields := make(map[string]string)
+	for _, key := range []string{"database", "table", "zookeeper_path", "replica_name", "last_queue_update_exception", "zookeeper_exception"} {
+		value, err := text(row, key)
+		if err != nil {
+			return replicaObservation{}, err
+		}
+		fields[key] = value
+	}
+	counts, err := numbers(row, "is_readonly", "is_session_expired", "total_replicas", "active_replicas",
+		"log_max_index", "log_pointer", "pending_data_tasks", "absolute_delay")
+	if err != nil {
+		return replicaObservation{}, err
+	}
+	if fields["zookeeper_path"] == "" || fields["replica_name"] == "" || fields["database"] == "" || fields["table"] == "" {
+		return replicaObservation{}, fmt.Errorf("missing replicated-table identity")
+	}
+	observation := replicaObservation{path: fields["zookeeper_path"], name: fields["replica_name"], pod: pod, total: counts["total_replicas"]}
+	// ClickHouse retains this exception even after subsequent queue updates succeed.
+	observation.historicalQueueErr = fields["last_queue_update_exception"] != ""
+	if observation.total < minReplicas || observation.total != int64(expected) || counts["active_replicas"] != observation.total {
+		observation.problems = append(observation.problems, fmt.Sprintf("%d/%d replicas active; %d expected", counts["active_replicas"], observation.total, expected))
+	}
+	if counts["is_readonly"] != 0 || counts["is_session_expired"] != 0 {
+		observation.problems = append(observation.problems, "replica is read-only or coordination session expired")
+	}
+	if counts["log_pointer"] <= counts["log_max_index"] || counts["pending_data_tasks"] != 0 {
+		observation.problems = append(observation.problems, fmt.Sprintf("replication backlog: %d data tasks; reported delay %ds", counts["pending_data_tasks"], counts["absolute_delay"]))
+	}
+	if fields["zookeeper_exception"] != "" {
+		observation.problems = append(observation.problems, "coordination query reported an exception")
+	}
+	return observation, nil
+}
+
+func evaluateClickHouse(observations []replicaObservation) Result {
+	groups := make(map[string][]replicaObservation)
+	var problems []string
+	historicalErrors := 0
+	for _, observation := range observations {
+		if observation.historicalQueueErr {
+			historicalErrors++
+		}
+		groups[observation.path] = append(groups[observation.path], observation)
+		for _, problem := range observation.problems {
+			problems = append(problems, fmt.Sprintf("%s %s: %s", observation.pod, observation.path, problem))
+		}
+	}
+	if len(groups) == 0 {
+		return result("clickhouse", Unknown, "no replicated tables found")
+	}
+	for path, replicas := range groups {
+		names := make(map[string]bool)
+		pods := make(map[string]bool)
+		for _, replica := range replicas {
+			if names[replica.name] || pods[replica.pod] || replica.total != replicas[0].total {
+				return result("clickhouse", Unknown, "duplicate or inconsistent replica evidence for "+path)
+			}
+			names[replica.name], pods[replica.pod] = true, true
+		}
+		if int64(len(replicas)) != replicas[0].total {
+			problems = append(problems, fmt.Sprintf("%s: queried %d/%d table replicas", path, len(replicas), replicas[0].total))
+		}
+	}
+	if len(problems) > 0 {
+		sort.Strings(problems)
+		return Result{Component: "clickhouse", Status: Degraded, Messages: problems}
+	}
+	report := result("clickhouse", Healthy, fmt.Sprintf("%d replicated table groups checked on every member; no pending data replication tasks", len(groups)))
+	if historicalErrors > 0 {
+		report.Messages = append(report.Messages, fmt.Sprintf("%d table replicas retain a previous queue-update exception; current replication checks pass", historicalErrors))
+	}
+	return report
+}
diff --git a/internal/orchestration/replication/discovery.go b/internal/orchestration/replication/discovery.go
new file mode 100644
index 0000000..4b04b3c
--- /dev/null
+++ b/internal/orchestration/replication/discovery.go
@@ -0,0 +1,128 @@
+package replication
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// Kubernetes is the read/query surface needed by the checker.
+type Kubernetes interface {
+	Clientset() kubernetes.Interface
+	Exec(context.Context, string, string, string, []string) ([]byte, error)
+}
+
+type inventory struct {
+	workloads []appsv1.StatefulSet
+	pods      []corev1.Pod
+}
+
+type member struct {
+	pod      corev1.Pod
+	replicas int
+}
+
+func (c *Checker) discover(ctx context.Context) (inventory, error) {
+	ctx, cancel := context.WithTimeout(ctx, c.options.RequestTimeout)
+	defer cancel()
+	options := metav1.ListOptions{LabelSelector: "app.kubernetes.io/name in (hbase,elasticsearch,kafka,clickhouse,zookeeper)"}
+	sets, err := c.kube.Clientset().AppsV1().StatefulSets(c.options.Namespace).List(ctx, options)
+	if err != nil {
+		return inventory{}, fmt.Errorf("list database StatefulSets: %w", err)
+	}
+	pods, err := c.kube.Clientset().CoreV1().Pods(c.options.Namespace).List(ctx, options)
+	if err != nil {
+		return inventory{}, fmt.Errorf("list database pods: %w", err)
+	}
+	return inventory{workloads: sets.Items, pods: pods.Items}, nil
+}
+
+func hasContainer(containers []corev1.Container, name string) bool {
+	return slices.ContainsFunc(containers, func(container corev1.Container) bool { return container.Name == name })
+}
+
+func (i inventory) members(name, component, container string) ([]member, error) {
+	var members []member
+	found := false
+	for _, workload := range i.workloads {
+		if workload.Labels["app.kubernetes.io/name"] != name ||
+			(component != "" && workload.Labels["app.kubernetes.io/component"] != component) {
+			continue
+		}
+		if !hasContainer(workload.Spec.Template.Spec.Containers, container) {
+			continue
+		}
+		found = true
+		pods, err := i.workloadPods(workload)
+		if err != nil {
+			return nil, err
+		}
+		for _, pod := range pods {
+			members = append(members, member{pod: pod, replicas: int(*workload.Spec.Replicas)})
+		}
+	}
+	if !found {
+		return nil, fmt.Errorf("no chart-managed StatefulSet for name=%q component=%q with container %q; check namespace, selected components and chart layout", name, component, container)
+	}
+	slices.SortFunc(members, func(a, b member) int {
+		if a.pod.Name < b.pod.Name {
+			return -1
+		}
+		if a.pod.Name > b.pod.Name {
+			return 1
+		}
+		return 0
+	})
+	return members, nil
+}
+
+func (i inventory) workloadPods(workload appsv1.StatefulSet) ([]corev1.Pod, error) {
+	if workload.Spec.Replicas == nil || *workload.Spec.Replicas < 1 {
+		return nil, fmt.Errorf("%s has no desired replicas", workload.Name)
+	}
+	expected := *workload.Spec.Replicas
+	if workload.DeletionTimestamp != nil || workload.Status.ObservedGeneration < workload.Generation ||
+		workload.Status.ReadyReplicas != expected || workload.Status.CurrentRevision != workload.Status.UpdateRevision {
+		return nil, fmt.Errorf("%s is not converged: %d/%d replicas ready", workload.Name, workload.Status.ReadyReplicas, expected)
+	}
+	var pods []corev1.Pod
+	for _, pod := range i.pods {
+		owner := metav1.GetControllerOf(&pod)
+		if owner == nil || owner.UID != workload.UID || owner.Kind != "StatefulSet" {
+			continue
+		}
+		if pod.DeletionTimestamp != nil || pod.Spec.NodeName == "" || !podReady(pod) {
+			return nil, fmt.Errorf("%s is terminating, unscheduled or not Ready", pod.Name)
+		}
+		pods = append(pods, pod)
+	}
+	if len(pods) != int(expected) {
+		return nil, fmt.Errorf("%s has %d/%d current pods", workload.Name, len(pods), expected)
+	}
+	return pods, nil
+}
+
+func podReady(pod corev1.Pod) bool {
+	return pod.Status.Phase == corev1.PodRunning && slices.ContainsFunc(pod.Status.Conditions, func(condition corev1.PodCondition) bool {
+		return condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue
+	})
+}
+
+func (i inventory) fingerprint() string {
+	var entries []string
+	for _, workload := range i.workloads {
+		entries = append(entries, fmt.Sprintf("sts:%s:%s", workload.UID, workload.ResourceVersion))
+	}
+	for _, pod := range i.pods {
+		entries = append(entries, fmt.Sprintf("pod:%s:%s", pod.UID, pod.ResourceVersion))
+	}
+	slices.Sort(entries)
+	data, _ := json.Marshal(entries)
+	return string(data)
+}
diff --git a/internal/orchestration/replication/elasticsearch.go b/internal/orchestration/replication/elasticsearch.go
new file mode 100644
index 0000000..3cc1d50
--- /dev/null
+++ b/internal/orchestration/replication/elasticsearch.go
@@ -0,0 +1,78 @@
+package replication
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+)
+
+const elasticsearchQuery = `scheme="$1"
+ca="$2"
+host="$3"
+set -- --fail --silent --show-error --max-time 20
+if [ -n "${ELASTIC_PASSWORD:-}" ]; then
+  set -- "$@" --user "elastic:${ELASTIC_PASSWORD}"
+fi
+if [ -n "$ca" ]; then
+  set -- "$@" --cacert "$ca"
+fi
+exec curl "$@" --resolve "${host}:9200:127.0.0.1" "${scheme}://${host}:9200/_cluster/health?level=indices"`
+
+func (c *Checker) checkElasticsearch(ctx context.Context, inventory inventory) Result {
+	members, err := inventory.members("elasticsearch", "", "elasticsearch")
+	if err != nil {
+		return result("elasticsearch", Unknown, err.Error())
+	}
+	if err := expectedMembers(members); err != nil {
+		return result("elasticsearch", Degraded, err.Error())
+	}
+	command := []string{"bash", "-ec", elasticsearchQuery, "replication-check", c.options.ElasticsearchScheme, c.options.ElasticsearchCA, c.options.ElasticsearchHost}
+	data, err := c.query(ctx, members[0].pod.Name, "elasticsearch", command)
+	if err != nil {
+		return result("elasticsearch", Unknown, err.Error())
+	}
+	return evaluateElasticsearch(data, len(members))
+}
+
+func evaluateElasticsearch(data []byte, expected int) Result {
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(data, &fields); err != nil {
+		return result("elasticsearch", Unknown, "invalid cluster health response")
+	}
+	counts, err := numbers(fields, "number_of_nodes", "unassigned_shards", "initializing_shards", "relocating_shards")
+	if err != nil {
+		return result("elasticsearch", Unknown, err.Error())
+	}
+	status, err := text(fields, "status")
+	if err != nil {
+		return result("elasticsearch", Unknown, err.Error())
+	}
+	var indices map[string]map[string]json.RawMessage
+	if err := json.Unmarshal(fields["indices"], &indices); err != nil || len(indices) == 0 {
+		return result("elasticsearch", Unknown, "no index replication evidence returned")
+	}
+	var problems []string
+	if status != "green" || counts["number_of_nodes"] != int64(expected) {
+		problems = append(problems, fmt.Sprintf("cluster status %s; %d/%d expected nodes", status, counts["number_of_nodes"], expected))
+	}
+	for _, key := range []string{"unassigned_shards", "initializing_shards", "relocating_shards"} {
+		if counts[key] > 0 {
+			problems = append(problems, fmt.Sprintf("%s=%d", key, counts[key]))
+		}
+	}
+	for name, index := range indices {
+		replicas, err := number(index, "number_of_replicas")
+		if err != nil {
+			return result("elasticsearch", Unknown, fmt.Sprintf("index %s: %v", name, err))
+		}
+		if replicas < minReplicas-1 {
+			problems = append(problems, fmt.Sprintf("index %s has no replica shard", name))
+		}
+	}
+	if len(problems) > 0 {
+		sort.Strings(problems)
+		return Result{Component: "elasticsearch", Status: Degraded, Messages: problems}
+	}
+	return result("elasticsearch", Healthy, fmt.Sprintf("%d indices have replica shards; all shards allocated with no recovery or relocation", len(indices)))
+}
diff --git a/internal/orchestration/replication/evaluators_test.go b/internal/orchestration/replication/evaluators_test.go
new file mode 100644
index 0000000..54fd6b0
--- /dev/null
+++ b/internal/orchestration/replication/evaluators_test.go
@@ -0,0 +1,186 @@
+package replication
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func encode(t *testing.T, value any) []byte {
+	t.Helper()
+	data, err := json.Marshal(value)
+	require.NoError(t, err)
+	return data
+}
+
+func hdfsFixture(t *testing.T, mutate func(map[string]any)) []byte {
+	t.Helper()
+	fields := map[string]any{
+		"name":                  "Hadoop:service=NameNode,name=FSNamesystem",
+		"UnderReplicatedBlocks": 0, "MissingBlocks": 0, "CorruptBlocks": 0, "PendingReplicationBlocks": 0, "NumLiveDataNodes": 3, "Safemode": "",
+	}
+	mutate(fields)
+	return encode(t, map[string]any{"configuredReplication": 3, "jmx": map[string]any{"beans": []any{fields}}})
+}
+
+func TestHDFSEvidence(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(map[string]any)
+		status string
+	}{
+		{"recovered", func(map[string]any) {}, Healthy},
+		{"replicating", func(m map[string]any) { m["UnderReplicatedBlocks"] = 12 }, Degraded},
+		{"missing block", func(m map[string]any) { m["MissingBlocks"] = 1 }, Degraded},
+		{"corrupt block", func(m map[string]any) { m["CorruptBlocks"] = 1 }, Degraded},
+		{"pending replication", func(m map[string]any) { m["PendingReplicationBlocks"] = 2 }, Degraded},
+		{"missing datanode", func(m map[string]any) { m["NumLiveDataNodes"] = 2 }, Degraded},
+		{"safe mode", func(m map[string]any) { m["Safemode"] = "ON" }, Degraded},
+		{"missing metric", func(m map[string]any) { delete(m, "UnderReplicatedBlocks") }, Unknown},
+		{"null metric", func(m map[string]any) { m["CorruptBlocks"] = nil }, Unknown},
+		{"negative metric", func(m map[string]any) { m["MissingBlocks"] = -1 }, Unknown},
+		{"unrelated bean", func(m map[string]any) { m["name"] = "Hadoop:service=Other" }, Unknown},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.status, evaluateHDFS(hdfsFixture(t, test.mutate), 3).Status)
+		})
+	}
+	data := strings.Replace(string(hdfsFixture(t, func(map[string]any) {})), `"configuredReplication":3`, `"configuredReplication":1`, 1)
+	assert.Equal(t, Degraded, evaluateHDFS([]byte(data), 3).Status)
+	assert.Equal(t, Unknown, evaluateHDFS([]byte(`{"beans":[]}`), 3).Status)
+}
+
+func esFixture() map[string]any {
+	return map[string]any{
+		"status": "green", "number_of_nodes": 3, "unassigned_shards": 0, "initializing_shards": 0, "relocating_shards": 0,
+		"indices": map[string]any{"events": map[string]any{"number_of_replicas": 1}},
+	}
+}
+
+func TestElasticsearchEvidence(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(map[string]any)
+		status string
+	}{
+		{"recovered", func(map[string]any) {}, Healthy},
+		{"yellow", func(m map[string]any) { m["status"] = "yellow" }, Degraded},
+		{"node missing", func(m map[string]any) { m["number_of_nodes"] = 2 }, Degraded},
+		{"initializing", func(m map[string]any) { m["initializing_shards"] = 1 }, Degraded},
+		{"relocating", func(m map[string]any) { m["relocating_shards"] = 1 }, Degraded},
+		{"unassigned", func(m map[string]any) { m["unassigned_shards"] = 1 }, Degraded},
+		{"green without replicas", func(m map[string]any) {
+			m["indices"] = map[string]any{"events": map[string]any{"number_of_replicas": 0}}
+		}, Degraded},
+		{"empty cluster", func(m map[string]any) { m["indices"] = map[string]any{} }, Unknown},
+		{"partial response", func(m map[string]any) { delete(m, "initializing_shards") }, Unknown},
+		{"missing index replication", func(m map[string]any) { m["indices"] = map[string]any{"events": map[string]any{}} }, Unknown},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fields := esFixture()
+			test.mutate(fields)
+			assert.Equal(t, test.status, evaluateElasticsearch(encode(t, fields), 3).Status)
+		})
+	}
+}
+
+func kafkaFixture() string {
+	var lines []string
+	for _, topic := range []string{"events", "__consumer_offsets", "__transaction_state"} {
+		lines = append(lines, fmt.Sprintf("Topic: %s TopicId: x PartitionCount: 1 ReplicationFactor: 2 Configs: min.insync.replicas=1", topic))
+		lines = append(lines, fmt.Sprintf("  Topic: %s Partition: 0 Leader: 0 Replicas: 0,1 Isr: 1,0", topic))
+	}
+	return strings.Join(lines, "\n")
+}
+
+func TestKafkaEvidence(t *testing.T) {
+	tests := []struct {
+		name, from, to, status string
+	}{
+		{"recovered", "Isr: 1,0", "Isr: 1,0", Healthy},
+		{"single replica transaction state", "Leader: 0 Replicas: 0,1 Isr: 1,0", "Leader: 0 Replicas: 0 Isr: 0", Degraded},
+		{"missing ISR", "Isr: 1,0", "Isr: 0", Degraded},
+		{"wrong ISR", "Isr: 1,0", "Isr: 2,0", Degraded},
+		{"missing leader", "Leader: 0", "Leader: -1", Degraded},
+		{"duplicate replica", "Replicas: 0,1", "Replicas: 0,0", Degraded},
+		{"empty ISR", "Isr: 1,0", "Isr:", Degraded},
+		{"missing internal topic", "__transaction_state", "another-topic", Unknown},
+		{"incomplete partitions", "PartitionCount: 1", "PartitionCount: 2", Unknown},
+		{"wrong partition ID", "Partition: 0", "Partition: 5", Unknown},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			output := strings.ReplaceAll(kafkaFixture(), test.from, test.to)
+			assert.Equal(t, test.status, evaluateKafka(output).Status)
+		})
+	}
+	assert.Equal(t, Unknown, evaluateKafka("").Status)
+	assert.Equal(t, Unknown, evaluateKafka(kafkaFixture()+"\n"+kafkaFixture()).Status)
+}
+
+func clickhouseFixture() map[string]any {
+	return map[string]any{
+		"database": "otel", "table": "traces", "zookeeper_path": "/clickhouse/tables/shard0/traces",
+		"replica_name": "replica0", "is_readonly": 0, "is_session_expired": 0,
+		"total_replicas": "2", "active_replicas": "2", "log_max_index": "100", "log_pointer": "101",
+		"pending_data_tasks": "0", "absolute_delay": "0", "last_queue_update_exception": "", "zookeeper_exception": "",
+	}
+}
+
+func TestClickHouseReplicaEvidence(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(map[string]any)
+		status string
+	}{
+		{"recovered", func(map[string]any) {}, Healthy},
+		{"merge backlog alone", func(m map[string]any) { m["queue_size"] = 8; m["merges_in_queue"] = 8 }, Healthy},
+		{"lagging log", func(m map[string]any) { m["log_pointer"] = "100" }, Degraded},
+		{"data backlog", func(m map[string]any) { m["pending_data_tasks"] = "1" }, Degraded},
+		{"read only", func(m map[string]any) { m["is_readonly"] = 1 }, Degraded},
+		{"expired session", func(m map[string]any) { m["is_session_expired"] = 1 }, Degraded},
+		{"inactive replica", func(m map[string]any) { m["active_replicas"] = "1" }, Degraded},
+		{"wrong replication", func(m map[string]any) { m["total_replicas"] = "1" }, Degraded},
+		{"coordination exception", func(m map[string]any) { m["zookeeper_exception"] = "connection failed" }, Degraded},
+		{"historical queue exception after recovery", func(m map[string]any) { m["last_queue_update_exception"] = "previous Keeper error" }, Healthy},
+		{"queue exception with backlog", func(m map[string]any) {
+			m["last_queue_update_exception"] = "Keeper error"
+			m["log_pointer"] = "100"
+		}, Degraded},
+		{"missing backlog evidence", func(m map[string]any) { delete(m, "pending_data_tasks") }, Unknown},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			row := clickhouseFixture()
+			test.mutate(row)
+			observations, err := parseClickHouse(encode(t, map[string]any{"data": []any{row}}), "pod0", 2)
+			if test.status == Unknown {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			peer := observations[0]
+			peer.pod, peer.name, peer.problems = "pod1", "replica1", nil
+			report := evaluateClickHouse(append(observations, peer))
+			assert.Equal(t, test.status, report.Status)
+			if test.name == "historical queue exception after recovery" {
+				assert.Contains(t, strings.Join(report.Messages, " "), "previous queue-update exception")
+			}
+		})
+	}
+}
+
+func TestClickHouseChecksEveryTableReplica(t *testing.T) {
+	observations, err := parseClickHouse(encode(t, map[string]any{"data": []any{clickhouseFixture()}}), "pod0", 2)
+	require.NoError(t, err)
+	assert.Equal(t, Degraded, evaluateClickHouse(observations).Status)
+	assert.Equal(t, Unknown, evaluateClickHouse(append(observations, observations...)).Status)
+	_, err = parseClickHouse([]byte(`{"data":[]}`), "pod0", 2)
+	require.Error(t, err)
+}
diff --git a/internal/orchestration/replication/hdfs.go b/internal/orchestration/replication/hdfs.go
new file mode 100644
index 0000000..1472950
--- /dev/null
+++ b/internal/orchestration/replication/hdfs.go
@@ -0,0 +1,85 @@
+package replication
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+)
+
+const hdfsQuery = `unset HADOOP_OPTS
+printf '{"configuredReplication":'
+hdfs getconf -confKey dfs.replication
+printf ',"jmx":'
+curl --fail --silent --show-error --max-time 20 'http://127.0.0.1:50070/jmx'
+printf '}'`
+
+func (c *Checker) checkHDFS(ctx context.Context, inventory inventory) Result {
+	namenodes, err := inventory.members("hbase", "hdfs-nn", "namenode")
+	if err != nil {
+		return result("hdfs", Unknown, err.Error())
+	}
+	if len(namenodes) != 1 {
+		return result("hdfs", Unknown, "expected the chart's single NameNode; external or HA NameNode layouts are not supported")
+	}
+	datanodes, err := inventory.members("hbase", "hdfs-dn", "datanode")
+	if err != nil {
+		return result("hdfs", Unknown, err.Error())
+	}
+	if err := expectedMembers(datanodes); err != nil {
+		return result("hdfs", Degraded, err.Error())
+	}
+	data, err := c.query(ctx, namenodes[0].pod.Name, "namenode", []string{"bash", "-ec", hdfsQuery})
+	if err != nil {
+		return result("hdfs", Unknown, err.Error())
+	}
+	return evaluateHDFS(data, len(datanodes))
+}
+
+func evaluateHDFS(data []byte, expected int) Result {
+	var response struct {
+		Replication *int `json:"configuredReplication"`
+		JMX         struct {
+			Beans []map[string]json.RawMessage `json:"beans"`
+		} `json:"jmx"`
+	}
+	if err := json.Unmarshal(data, &response); err != nil || response.Replication == nil {
+		return result("hdfs", Unknown, "invalid HDFS replication/JMX response")
+	}
+	fields := make(map[string]json.RawMessage)
+	for _, bean := range response.JMX.Beans {
+		name, _ := text(bean, "name")
+		switch name {
+		case "Hadoop:service=NameNode,name=FSNamesystem", "Hadoop:service=NameNode,name=FSNamesystemState", "Hadoop:service=NameNode,name=NameNodeInfo":
+			for key, value := range bean {
+				fields[key] = value
+			}
+		}
+	}
+	counts, err := numbers(fields, "UnderReplicatedBlocks", "MissingBlocks", "CorruptBlocks", "PendingReplicationBlocks", "NumLiveDataNodes")
+	if err != nil {
+		return result("hdfs", Unknown, err.Error())
+	}
+	safemode, err := text(fields, "Safemode")
+	if err != nil {
+		return result("hdfs", Unknown, err.Error())
+	}
+	var problems []string
+	if *response.Replication < minReplicas {
+		problems = append(problems, fmt.Sprintf("configured block replication is %d; at least %d required", *response.Replication, minReplicas))
+	}
+	if safemode != "" {
+		problems = append(problems, "NameNode is in safe mode")
+	}
+	if counts["NumLiveDataNodes"] != int64(expected) {
+		problems = append(problems, fmt.Sprintf("%d/%d expected DataNodes are live", counts["NumLiveDataNodes"], expected))
+	}
+	for _, key := range []string{"UnderReplicatedBlocks", "MissingBlocks", "CorruptBlocks", "PendingReplicationBlocks"} {
+		if counts[key] != 0 {
+			problems = append(problems, fmt.Sprintf("%s=%d", key, counts[key]))
+		}
+	}
+	if len(problems) > 0 {
+		return Result{Component: "hdfs", Status: Degraded, Messages: problems}
+	}
+	return result("hdfs", Healthy, fmt.Sprintf("%d DataNodes live; no missing, corrupt, under-replicated or pending-replication blocks", expected))
+}
diff --git a/internal/orchestration/replication/json.go b/internal/orchestration/replication/json.go
new file mode 100644
index 0000000..ee6b998
--- /dev/null
+++ b/internal/orchestration/replication/json.go
@@ -0,0 +1,54 @@
+package replication
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+)
+
+func number(fields map[string]json.RawMessage, key string) (int64, error) {
+	raw, found := fields[key]
+	if !found || string(raw) == "null" {
+		return 0, fmt.Errorf("missing numeric field %s", key)
+	}
+	var value int64
+	if err := json.Unmarshal(raw, &value); err != nil {
+		var quoted string
+		if err := json.Unmarshal(raw, &quoted); err != nil {
+			return 0, fmt.Errorf("invalid integer field %s", key)
+		}
+		var parseErr error
+		value, parseErr = strconv.ParseInt(quoted, 10, 64)
+		if parseErr != nil {
+			return 0, fmt.Errorf("invalid integer field %s", key)
+		}
+	}
+	if value < 0 {
+		return 0, fmt.Errorf("invalid nonnegative integer field %s", key)
+	}
+	return value, nil
+}
+
+func text(fields map[string]json.RawMessage, key string) (string, error) {
+	raw, found := fields[key]
+	if !found || string(raw) == "null" {
+		return "", fmt.Errorf("missing string field %s", key)
+	}
+	var value string
+	if err := json.Unmarshal(raw, &value); err != nil {
+		return "", fmt.Errorf("invalid string field %s", key)
+	}
+	return value, nil
+}
+
+func numbers(fields map[string]json.RawMessage, keys ...string) (map[string]int64, error) {
+	values := make(map[string]int64, len(keys))
+	for _, key := range keys {
+		value, err := number(fields, key)
+		if err != nil {
+			return nil, err
+		}
+		values[key] = value
+	}
+	return values, nil
+}
diff --git a/internal/orchestration/replication/kafka.go b/internal/orchestration/replication/kafka.go
new file mode 100644
index 0000000..1880613
--- /dev/null
+++ b/internal/orchestration/replication/kafka.go
@@ -0,0 +1,141 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// The CLI must not bind the broker's inherited JMX port.
+const kafkaQuery = `unset JMX_PORT KAFKA_JMX_OPTS
+exec kafka-topics.sh "$@"`
+
+var (
+	kafkaTopicPattern     = regexp.MustCompile(`\bTopic:\s*(\S+)`)
+	kafkaPartitionPattern = regexp.MustCompile(`\bPartition:\s*(\d+)\b`)
+	kafkaCountPattern     = regexp.MustCompile(`\bPartitionCount:\s*(\d+)\b`)
+	kafkaFieldsPattern    = regexp.MustCompile(`\b(Leader|Replicas|Isr):\s*([-\d,]+)`)
+)
+
+func (c *Checker) checkKafka(ctx context.Context, inventory inventory) Result {
+	members, err := inventory.members("kafka", "kafka", "kafka")
+	if err != nil {
+		return result("kafka", Unknown, err.Error())
+	}
+	if err := expectedMembers(members); err != nil {
+		return result("kafka", Degraded, err.Error())
+	}
+	command := []string{"bash", "-ec", kafkaQuery, "replication-check", "--bootstrap-server", c.options.KafkaBootstrapServer, "--describe"}
+	if c.options.KafkaClientProperties != "" {
+		command = append(command, "--command-config", c.options.KafkaClientProperties)
+	}
+	data, err := c.query(ctx, members[0].pod.Name, "kafka", command)
+	if err != nil {
+		return result("kafka", Unknown, err.Error())
+	}
+	return evaluateKafka(string(data))
+}
+
+func brokerIDs(value string) (map[int]bool, error) {
+	if value == "" {
+		return nil, fmt.Errorf("missing broker IDs")
+	}
+	ids := make(map[int]bool)
+	for _, part := range strings.Split(value, ",") {
+		id, err := strconv.Atoi(part)
+		if err != nil || id < 0 || ids[id] {
+			return nil, fmt.Errorf("invalid or duplicate broker ID")
+		}
+		ids[id] = true
+	}
+	return ids, nil
+}
+
+func partitionProblem(line string) string {
+	fields := make(map[string]string)
+	for _, match := range kafkaFieldsPattern.FindAllStringSubmatch(line, -1) {
+		fields[match[1]] = match[2]
+	}
+	replicas, err := brokerIDs(fields["Replicas"])
+	if err != nil || len(replicas) < minReplicas {
+		return "fewer than two distinct assigned replicas or invalid assignment"
+	}
+	isr, err := brokerIDs(fields["Isr"])
+	if err != nil || len(isr) != len(replicas) {
+		return "assigned replicas are not all in sync"
+	}
+	for id := range replicas {
+		if !isr[id] {
+			return "assigned replicas are not all in sync"
+		}
+	}
+	leader, err := strconv.Atoi(fields["Leader"])
+	if err != nil || !isr[leader] {
+		return "no available in-sync leader"
+	}
+	return ""
+}
+
+func evaluateKafka(output string) Result {
+	counts := make(map[string]int)
+	partitions := make(map[string]map[int]bool)
+	var problems []string
+	for _, line := range strings.Split(output, "\n") {
+		topicMatch := kafkaTopicPattern.FindStringSubmatch(line)
+		if topicMatch == nil {
+			continue
+		}
+		topic := topicMatch[1]
+		if count := kafkaCountPattern.FindStringSubmatch(line); count != nil {
+			value, err := strconv.Atoi(count[1])
+			if err != nil || value < 1 || counts[topic] != 0 {
+				return result("kafka", Unknown, "invalid or duplicate topic summary")
+			}
+			counts[topic] = value
+		}
+		if partition := kafkaPartitionPattern.FindStringSubmatch(line); partition != nil {
+			id, err := strconv.Atoi(partition[1])
+			if err != nil || partitions[topic][id] {
+				return result("kafka", Unknown, "invalid or duplicate partition description")
+			}
+			if partitions[topic] == nil {
+				partitions[topic] = make(map[int]bool)
+			}
+			partitions[topic][id] = true
+			if problem := partitionProblem(line); problem != "" {
+				problems = append(problems, fmt.Sprintf("%s partition %d: %s", topic, id, problem))
+			}
+		}
+	}
+	if err := completeKafkaEvidence(counts, partitions); err != nil {
+		return result("kafka", Unknown, err.Error())
+	}
+	if len(problems) > 0 {
+		return Result{Component: "kafka", Status: Degraded, Messages: problems}
+	}
+	return result("kafka", Healthy, fmt.Sprintf("all partitions of %d topics have at least two replicas, complete ISR and an in-sync leader", len(counts)))
+}
+
+func completeKafkaEvidence(counts map[string]int, partitions map[string]map[int]bool) error {
+	for _, topic := range []string{"__consumer_offsets", "__transaction_state"} {
+		if counts[topic] == 0 {
+			return fmt.Errorf("required internal topic %s is absent; initialize its workload and repeat the check", topic)
+		}
+	}
+	if len(counts) != len(partitions) {
+		return fmt.Errorf("topic summaries and partition descriptions do not match")
+	}
+	for topic, count := range counts {
+		if len(partitions[topic]) != count {
+			return fmt.Errorf("incomplete partition descriptions for %s", topic)
+		}
+		for id := 0; id < count; id++ {
+			if !partitions[topic][id] {
+				return fmt.Errorf("missing partition %d for %s", id, topic)
+			}
+		}
+	}
+	return nil
+}
diff --git a/internal/orchestration/replication/kafka_test.go b/internal/orchestration/replication/kafka_test.go
new file mode 100644
index 0000000..8777445
--- /dev/null
+++ b/internal/orchestration/replication/kafka_test.go
@@ -0,0 +1,40 @@
+package replication
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestKafkaQueryDoesNotInheritBrokerJMXPort(t *testing.T) {
+	bash, err := exec.LookPath("bash")
+	if err != nil {
+		t.Skip("bash is required to exercise the in-pod query")
+	}
+	dir := t.TempDir()
+	script := `#!/bin/sh
+test -z "${JMX_PORT:-}" || exit 1
+test -z "${KAFKA_JMX_OPTS:-}" || exit 1
+printf '%s\n' "$KAFKA_OPTS" "$@"
+`
+	path := filepath.Join(dir, "kafka-topics.sh")
+	require.NoError(t, os.WriteFile(path, []byte(script), 0o600))
+	require.NoError(t, os.Chmod(path, 0o700))
+	t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("JMX_PORT", "5555")
+	t.Setenv("KAFKA_JMX_OPTS", "-Dcom.sun.management.jmxremote.port=5555")
+	t.Setenv("KAFKA_OPTS", "-Djava.security.auth.login.config=/mounted/jaas.conf")
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	command := exec.CommandContext(ctx, bash, "-ec", kafkaQuery, "replication-check",
+		"--bootstrap-server", "localhost:9092", "--describe", "--command-config", "/mounted/client properties")
+	output, err := command.CombinedOutput()
+	require.NoError(t, err, string(output))
+	assert.Equal(t, "-Djava.security.auth.login.config=/mounted/jaas.conf\n--bootstrap-server\nlocalhost:9092\n--describe\n--command-config\n/mounted/client properties\n", string(output))
+}
diff --git a/internal/orchestration/replication/report.go b/internal/orchestration/replication/report.go
new file mode 100644
index 0000000..b3aad2d
--- /dev/null
+++ b/internal/orchestration/replication/report.go
@@ -0,0 +1,88 @@
+// Package replication checks the observed replication state of chart-managed databases.
+// It does not grant permission to disrupt a node or modify database state.
+package replication
+
+import (
+	"fmt"
+	"slices"
+	"time"
+)
+
+const (
+	Healthy  = "healthy"
+	Degraded = "degraded"
+	Unknown  = "unknown"
+
+	minReplicas = 2
+)
+
+var supportedComponents = []string{"hdfs", "elasticsearch", "kafka", "clickhouse", "zookeeper"}
+
+// Options identifies the installation and bounds each database query.
+type Options struct {
+	Namespace             string
+	Components            []string
+	RequestTimeout        time.Duration
+	KafkaClientProperties string
+	KafkaBootstrapServer  string
+	ElasticsearchScheme   string
+	ElasticsearchCA       string
+	ElasticsearchHost     string
+}
+
+// Validate rejects ambiguous scope and unsupported probe settings.
+func (o Options) Validate() error {
+	if o.Namespace == "" {
+		return fmt.Errorf("namespace is required")
+	}
+	if o.RequestTimeout <= 0 {
+		return fmt.Errorf("request-timeout must be positive")
+	}
+	if o.ElasticsearchScheme != "http" && o.ElasticsearchScheme != "https" {
+		return fmt.Errorf("elasticsearch-scheme must be http or https")
+	}
+	if len(o.Components) == 0 {
+		return fmt.Errorf("select at least one component")
+	}
+	seen := make(map[string]bool)
+	for _, component := range o.Components {
+		if !slices.Contains(supportedComponents, component) || seen[component] {
+			return fmt.Errorf("invalid or duplicate component %q; choose from %v", component, supportedComponents)
+		}
+		seen[component] = true
+	}
+	return nil
+}
+
+// Result reports one component; missing evidence is unknown, never healthy.
+type Result struct {
+	Component string   `json:"component"`
+	Status    string   `json:"status"`
+	Messages  []string `json:"messages"`
+}
+
+// Report is a point-in-time observation, not a maintenance lock.
+type Report struct {
+	CheckedAt time.Time `json:"checkedAt"`
+	Namespace string    `json:"namespace"`
+	Status    string    `json:"status"`
+	Error     string    `json:"error,omitempty"`
+	Checks    []Result  `json:"checks"`
+}
+
+func result(component, status, message string) Result {
+	return Result{Component: component, Status: status, Messages: []string{message}}
+}
+
+func reportStatus(checks []Result) string {
+	status := Healthy
+	for _, check := range checks {
+		if check.Status == Unknown {
+			return Unknown
+		}
+		if check.Status != Healthy {
+			status = Degraded
+		}
+	}
+	return status
+}
diff --git a/internal/orchestration/replication/wait.go b/internal/orchestration/replication/wait.go
new file mode 100644
index 0000000..a03e9f4
--- /dev/null
+++ b/internal/orchestration/replication/wait.go
@@ -0,0 +1,71 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// WaitProgress describes the stability period after a completed observation.
+type WaitProgress struct {
+	ObservedAt time.Time
+	HealthyFor time.Duration
+	Required   time.Duration
+	Reset      bool
+	Complete   bool
+}
+
+// Wait polls until replication remains healthy for stableFor or the caller's deadline expires.
+// Interrupted observations never replace the last completed report.
+func Wait(ctx context.Context, check func(context.Context) Report, interval, stableFor time.Duration, observe func(Report, WaitProgress)) (Report, error) {
+	if interval <= 0 || stableFor < 0 {
+		return Report{}, fmt.Errorf("interval must be positive and stable-for cannot be negative")
+	}
+	var healthySince time.Time
+	var report Report
+	for {
+		if err := ctx.Err(); err != nil {
+			return report, fmt.Errorf("replication wait ended: %w", err)
+		}
+		next := check(ctx)
+		if err := ctx.Err(); err != nil {
+			if report.Namespace == "" {
+				report.Namespace = next.Namespace
+			}
+			return report, fmt.Errorf("replication wait ended: %w", err)
+		}
+		report = next
+		now := time.Now()
+		progress := WaitProgress{ObservedAt: now.UTC(), Required: stableFor}
+		if report.Status != Healthy {
+			progress.Reset = !healthySince.IsZero()
+			healthySince = time.Time{}
+		} else {
+			if healthySince.IsZero() {
+				healthySince = now
+			}
+			progress.HealthyFor = now.Sub(healthySince)
+			progress.Complete = progress.HealthyFor >= stableFor
+		}
+		if observe != nil {
+			observe(report, progress)
+		}
+		if err := ctx.Err(); err != nil {
+			return report, fmt.Errorf("replication wait ended: %w", err)
+		}
+		if progress.Complete {
+			return report, nil
+		}
+		delay := interval
+		if report.Status == Healthy {
+			delay = min(delay, stableFor-progress.HealthyFor)
+		}
+		timer := time.NewTimer(delay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return report, fmt.Errorf("replication wait ended: %w", ctx.Err())
+		case <-timer.C:
+		}
+	}
+}
diff --git a/internal/orchestration/replication/wait_test.go b/internal/orchestration/replication/wait_test.go
new file mode 100644
index 0000000..c11eb86
--- /dev/null
+++ b/internal/orchestration/replication/wait_test.go
@@ -0,0 +1,142 @@
+package replication
+
+import (
+	"context"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWaitContinuesThroughUnknownAndDegraded(t *testing.T) {
+	statuses := []string{Unknown, Degraded, Healthy}
+	calls := 0
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	report, err := Wait(ctx, func(context.Context) Report {
+		status := statuses[calls]
+		calls++
+		return Report{Status: status}
+	}, time.Millisecond, 0, nil)
+	require.NoError(t, err)
+	assert.Equal(t, Healthy, report.Status)
+	assert.Equal(t, 3, calls)
+}
+
+func TestWaitHonorsDeadlineDuringQuery(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
+	defer cancel()
+	_, err := Wait(ctx, func(ctx context.Context) Report {
+		<-ctx.Done()
+		return Report{Status: Healthy}
+	}, time.Millisecond, 0, nil)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+}
+
+func TestWaitRequiresSustainedHealthyObservations(t *testing.T) {
+	calls := 0
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_, err := Wait(ctx, func(context.Context) Report {
+		calls++
+		if calls == 2 {
+			return Report{Status: Degraded}
+		}
+		return Report{Status: Healthy}
+	}, time.Millisecond, 5*time.Millisecond, nil)
+	require.NoError(t, err)
+	assert.GreaterOrEqual(t, calls, 4)
+}
+
+func TestWaitCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	_, err := Wait(ctx, func(context.Context) Report {
+		cancel()
+		return Report{Status: Unknown}
+	}, time.Hour, 0, nil)
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+func TestWaitExitsAfterDefaultStabilityPeriodWithSlowChecks(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		start := time.Now()
+		var progress []WaitProgress
+		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+		defer cancel()
+		report, err := Wait(ctx, func(context.Context) Report {
+			time.Sleep(2 * time.Second)
+			return Report{Status: Healthy}
+		}, 10*time.Second, 30*time.Second, func(_ Report, state WaitProgress) {
+			progress = append(progress, state)
+		})
+		require.NoError(t, err)
+		assert.Equal(t, Healthy, report.Status)
+		require.Len(t, progress, 4)
+		assert.Equal(t, time.Duration(0), progress[0].HealthyFor)
+		assert.Equal(t, 12*time.Second, progress[1].HealthyFor)
+		assert.Equal(t, 32*time.Second, progress[3].HealthyFor)
+		assert.True(t, progress[3].Complete)
+		assert.Equal(t, 34*time.Second, time.Since(start))
+	})
+}
+
+func TestWaitRechecksAtStabilityDeadlineBeforeLongInterval(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		start := time.Now()
+		ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+		defer cancel()
+		calls := 0
+		_, err := Wait(ctx, func(context.Context) Report {
+			calls++
+			return Report{Status: Healthy}
+		}, time.Minute, 30*time.Second, nil)
+		require.NoError(t, err)
+		assert.Equal(t, 2, calls)
+		assert.Equal(t, 30*time.Second, time.Since(start))
+	})
+}
+
+func TestWaitResetsStabilityProgress(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		calls := 0
+		var progress []WaitProgress
+		_, err := Wait(context.Background(), func(context.Context) Report {
+			calls++
+			if calls == 3 {
+				return Report{Status: Degraded}
+			}
+			return Report{Status: Healthy}
+		}, 10*time.Second, 30*time.Second, func(_ Report, state WaitProgress) {
+			progress = append(progress, state)
+		})
+		require.NoError(t, err)
+		require.Len(t, progress, 7)
+		assert.True(t, progress[2].Reset)
+		assert.Zero(t, progress[3].HealthyFor)
+		assert.Equal(t, 30*time.Second, progress[6].HealthyFor)
+		assert.True(t, progress[6].Complete)
+	})
+}
+
+func TestWaitCancellationPreservesLastCompletedObservation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		calls, observed := 0, 0
+		lastCompleted := Report{Namespace: "test", CheckedAt: time.Now(), Status: Healthy,
+			Checks: []Result{result("kafka", Healthy, "all partitions in sync")}}
+		report, err := Wait(ctx, func(context.Context) Report {
+			calls++
+			if calls == 3 {
+				cancel()
+				return Report{Status: Unknown, Checks: []Result{result("kafka", Unknown, "aborted query URL")}}
+			}
+			return lastCompleted
+		}, 10*time.Second, 30*time.Second, func(Report, WaitProgress) { observed++ })
+		require.ErrorIs(t, err, context.Canceled)
+		assert.Equal(t, lastCompleted, report)
+		assert.Equal(t, 2, observed)
+	})
+}
diff --git a/internal/orchestration/replication/zookeeper.go b/internal/orchestration/replication/zookeeper.go
new file mode 100644
index 0000000..f04478d
--- /dev/null
+++ b/internal/orchestration/replication/zookeeper.go
@@ -0,0 +1,169 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+const (
+	minZooKeeperVoters = 3
+	zooKeeperLeader    = "leader"
+	zooKeeperFollower  = "follower"
+)
+
+const zookeeperQuery = `exec 3<>/dev/tcp/127.0.0.1/2181
+printf mntr >&3
+while true; do
+  if IFS= read -r -t 5 -u 3 line; then
+    printf '%s\n' "$line"
+  else
+    status=$?
+    if [ "$status" -eq 1 ] && [ -z "$line" ]; then
+      break
+    fi
+    exit 1
+  fi
+done`
+
+type zooKeeperObservation struct {
+	pod       string
+	role      string
+	peerState string
+	voters    int
+	synced    int
+}
+
+func (c *Checker) checkZooKeeper(ctx context.Context, inventory inventory) Result {
+	members, err := inventory.members("zookeeper", "zookeeper", "zookeeper")
+	if err != nil {
+		return result("zookeeper", Unknown, err.Error())
+	}
+	if len(members) < minZooKeeperVoters {
+		return result("zookeeper", Degraded, "at least three voting members are required for fault-tolerant ZooKeeper")
+	}
+	var observations []zooKeeperObservation
+	for _, member := range members {
+		if member.replicas != len(members) {
+			return result("zookeeper", Unknown, "expected one chart-managed ZooKeeper ensemble")
+		}
+		observation, err := c.queryZooKeeper(ctx, member.pod.Name)
+		if err != nil {
+			return result("zookeeper", Unknown, err.Error())
+		}
+		observations = append(observations, observation)
+	}
+	report := evaluateZooKeeper(observations)
+	if report.Status != Healthy {
+		return report
+	}
+	// Recheck the leader after sampling followers; elections do not change Pod status.
+	for n, observation := range observations {
+		if observation.role != zooKeeperLeader {
+			continue
+		}
+		current, err := c.queryZooKeeper(ctx, observation.pod)
+		if err != nil {
+			return result("zookeeper", Unknown, err.Error())
+		}
+		if current.role != zooKeeperLeader {
+			return result("zookeeper", Unknown, "ZooKeeper leadership changed during the checks; repeat the observation")
+		}
+		observations[n] = current
+	}
+	return evaluateZooKeeper(observations)
+}
+
+func (c *Checker) queryZooKeeper(ctx context.Context, pod string) (zooKeeperObservation, error) {
+	data, err := c.query(ctx, pod, "zookeeper", []string{"bash", "-ec", zookeeperQuery})
+	if err != nil {
+		return zooKeeperObservation{}, err
+	}
+	observation, err := parseZooKeeper(string(data), pod)
+	if err != nil {
+		return zooKeeperObservation{}, fmt.Errorf("%s: %w", pod, err)
+	}
+	return observation, nil
+}
+
+func parseZooKeeper(data, pod string) (zooKeeperObservation, error) {
+	fields := make(map[string]string)
+	for _, line := range strings.Split(data, "\n") {
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+		key, value, ok := strings.Cut(line, "\t")
+		_, duplicate := fields[key]
+		if !ok || key == "" || duplicate {
+			return zooKeeperObservation{}, fmt.Errorf("invalid or unavailable ZooKeeper mntr response")
+		}
+		fields[key] = strings.TrimSpace(value)
+	}
+	if fields["zk_server_state"] == "" {
+		return zooKeeperObservation{}, fmt.Errorf("missing zk_server_state in mntr response")
+	}
+	observation := zooKeeperObservation{pod: pod, role: fields["zk_server_state"], peerState: fields["zk_peer_state"]}
+	voters, err := zooKeeperNumber(fields, "zk_quorum_size")
+	if err != nil {
+		return zooKeeperObservation{}, err
+	}
+	observation.voters = voters
+	if observation.role == zooKeeperLeader {
+		observation.synced, err = zooKeeperNumber(fields, "zk_synced_followers")
+		if err != nil {
+			return zooKeeperObservation{}, err
+		}
+	}
+	return observation, nil
+}
+
+func zooKeeperNumber(fields map[string]string, key string) (int, error) {
+	value, err := strconv.Atoi(fields[key])
+	if err != nil || value < 0 {
+		return 0, fmt.Errorf("missing or invalid %s in mntr response", key)
+	}
+	return value, nil
+}
+
+func evaluateZooKeeper(observations []zooKeeperObservation) Result {
+	if len(observations) < minZooKeeperVoters {
+		return result("zookeeper", Degraded, "at least three voting members are required for fault-tolerant ZooKeeper")
+	}
+	leaders := 0
+	seen := make(map[string]bool)
+	var problems []string
+	for _, observation := range observations {
+		if observation.pod == "" || seen[observation.pod] {
+			return result("zookeeper", Unknown, "missing or duplicate ZooKeeper member evidence")
+		}
+		seen[observation.pod] = true
+		if observation.voters != len(observations) {
+			return result("zookeeper", Unknown, fmt.Sprintf("%s reports %d configured voters; %d members discovered", observation.pod, observation.voters, len(observations)))
+		}
+		switch observation.role {
+		case zooKeeperLeader:
+			leaders++
+			if observation.synced != len(observations)-1 {
+				problems = append(problems, fmt.Sprintf("%s reports %d/%d synchronized followers", observation.pod, observation.synced, len(observations)-1))
+			}
+		case zooKeeperFollower:
+		default:
+			problems = append(problems, fmt.Sprintf("%s is %s; expected a voting leader or follower", observation.pod, observation.role))
+		}
+		expectedPeerState := "following - broadcast"
+		if observation.role == zooKeeperLeader {
+			expectedPeerState = "leading - broadcast"
+		}
+		if observation.peerState != "" && observation.peerState != expectedPeerState {
+			problems = append(problems, fmt.Sprintf("%s is not in broadcast state: %s", observation.pod, observation.peerState))
+		}
+	}
+	if leaders != 1 {
+		problems = append(problems, fmt.Sprintf("expected one ZooKeeper leader; observed %d", leaders))
+	}
+	if len(problems) > 0 {
+		return Result{Component: "zookeeper", Status: Degraded, Messages: problems}
+	}
+	return result("zookeeper", Healthy, fmt.Sprintf("%d voting members available; one leader and %d synchronized followers", len(observations), len(observations)-1))
+}
diff --git a/internal/orchestration/replication/zookeeper_test.go b/internal/orchestration/replication/zookeeper_test.go
new file mode 100644
index 0000000..dfce640
--- /dev/null
+++ b/internal/orchestration/replication/zookeeper_test.go
@@ -0,0 +1,147 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func zooKeeperFixture(role string) string {
+	data := "zk_version\t3.9.5\nzk_server_state\t" + role + "\nzk_quorum_size\t3\nzk_avg_latency\t0.123\n"
+	if role == zooKeeperLeader {
+		return data + "zk_peer_state\tleading - broadcast\nzk_synced_followers\t2\n"
+	}
+	return data + "zk_peer_state\tfollowing - broadcast\nzk_synced_observers\tnull\n"
+}
+
+func TestZooKeeperParsing(t *testing.T) {
+	tests := []struct {
+		name, data string
+		valid      bool
+	}{
+		{zooKeeperLeader, zooKeeperFixture(zooKeeperLeader), true},
+		{zooKeeperFollower, zooKeeperFixture(zooKeeperFollower), true},
+		{"optional peer state absent", strings.ReplaceAll(zooKeeperFixture(zooKeeperFollower), "zk_peer_state\tfollowing - broadcast\n", ""), true},
+		{"disabled command", "mntr is not executed because it is not in the whitelist.\n", false},
+		{"not serving", "This ZooKeeper instance is not currently serving requests\n", false},
+		{"empty response", "", false},
+		{"missing role", "zk_quorum_size\t3\n", false},
+		{"missing membership", "zk_server_state\tfollower\n", false},
+		{"missing synchronized count", "zk_server_state\tleader\nzk_quorum_size\t3\n", false},
+		{"duplicate role", zooKeeperFixture(zooKeeperLeader) + "zk_server_state\tfollower\n", false},
+		{"duplicate empty value", "zk_version\t\n" + zooKeeperFixture(zooKeeperLeader), false},
+		{"negative synchronized count", strings.ReplaceAll(zooKeeperFixture(zooKeeperLeader), "zk_synced_followers\t2", "zk_synced_followers\t-1"), false},
+		{"invalid membership", strings.ReplaceAll(zooKeeperFixture(zooKeeperFollower), "zk_quorum_size\t3", "zk_quorum_size\tnull"), false},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			observation, err := parseZooKeeper(test.data, "zk-0")
+			if !test.valid {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, 3, observation.voters)
+			assert.Equal(t, "zk-0", observation.pod)
+		})
+	}
+}
+
+func TestZooKeeperRequiresFullVotingEnsemble(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func([]zooKeeperObservation) []zooKeeperObservation
+		status string
+	}{
+		{"healthy", func(o []zooKeeperObservation) []zooKeeperObservation { return o }, Healthy},
+		{"quorum without full recovery", func(o []zooKeeperObservation) []zooKeeperObservation { o[0].synced = 1; return o }, Degraded},
+		{"no leader", func(o []zooKeeperObservation) []zooKeeperObservation { o[0].role = zooKeeperFollower; return o }, Degraded},
+		{"two leaders", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = zooKeeperLeader; return o }, Degraded},
+		{"election", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "looking"; return o }, Degraded},
+		{"observer", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "observer"; return o }, Degraded},
+		{"standalone", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].role = "standalone"; return o }, Degraded},
+		{"synchronizing", func(o []zooKeeperObservation) []zooKeeperObservation {
+			o[1].peerState = "following - synchronization"
+			return o
+		}, Degraded},
+		{"inconsistent role", func(o []zooKeeperObservation) []zooKeeperObservation {
+			o[0].peerState = "following - broadcast"
+			return o
+		}, Degraded},
+		{"unexpected voter configuration", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].voters = 5; return o }, Unknown},
+		{"duplicate member", func(o []zooKeeperObservation) []zooKeeperObservation { o[1].pod = o[0].pod; return o }, Unknown},
+		{"too few voters", func(o []zooKeeperObservation) []zooKeeperObservation { return o[:2] }, Degraded},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			observations := []zooKeeperObservation{
+				{pod: "zk-0", role: zooKeeperLeader, voters: 3, synced: 2},
+				{pod: "zk-1", role: zooKeeperFollower, voters: 3},
+				{pod: "zk-2", role: zooKeeperFollower, voters: 3},
+			}
+			assert.Equal(t, test.status, evaluateZooKeeper(test.mutate(observations)).Status)
+		})
+	}
+}
+
+func TestZooKeeperQueriesEveryMemberAndRechecksLeader(t *testing.T) {
+	tests := []struct {
+		name, status string
+		finalRole    string
+		finalSynced  string
+		queryError   bool
+	}{
+		{name: "healthy", status: Healthy, finalRole: zooKeeperLeader, finalSynced: "2"},
+		{name: "election during observation", status: Unknown, finalRole: zooKeeperFollower},
+		{name: "follower falls behind during observation", status: Degraded, finalRole: zooKeeperLeader, finalSynced: "1"},
+		{name: "query denied", status: Unknown, queryError: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var queried []string
+			kube := &fakeKubernetes{
+				client: fake.NewSimpleClientset(databaseObjects("zookeeper", "zookeeper", "zookeeper", 3)...),
+				exec: func(_ context.Context, namespace, pod, container string, command []string) ([]byte, error) {
+					assert.Equal(t, "test", namespace)
+					assert.Equal(t, "zookeeper", container)
+					assert.Equal(t, []string{"bash", "-ec", zookeeperQuery}, command)
+					queried = append(queried, pod)
+					if test.queryError {
+						return nil, fmt.Errorf("query denied")
+					}
+					if len(queried) == 4 {
+						return []byte(strings.ReplaceAll(zooKeeperFixture(test.finalRole), "zk_synced_followers\t2", "zk_synced_followers\t"+test.finalSynced)), nil
+					}
+					if pod == "zookeeper-0" {
+						return []byte(zooKeeperFixture(zooKeeperLeader)), nil
+					}
+					return []byte(zooKeeperFixture(zooKeeperFollower)), nil
+				},
+			}
+			options := testOptions()
+			options.Components = []string{"zookeeper"}
+			probe, err := New(kube, options)
+			require.NoError(t, err)
+			assert.Equal(t, test.status, probe.Check(context.Background()).Status)
+			if !test.queryError {
+				assert.Equal(t, []string{"zookeeper-0", "zookeeper-1", "zookeeper-2", "zookeeper-0"}, queried)
+			}
+		})
+	}
+}
+
+func TestZooKeeperMissingPodCannotPass(t *testing.T) {
+	objects := databaseObjects("zookeeper", "zookeeper", "zookeeper", 3)
+	kube := &fakeKubernetes{client: fake.NewSimpleClientset(objects[:len(objects)-1]...)}
+	options := testOptions()
+	options.Components = []string{"zookeeper"}
+	probe, err := New(kube, options)
+	require.NoError(t, err)
+	assert.Equal(t, Unknown, probe.Check(context.Background()).Status)
+	assert.Zero(t, kube.calls)
+}