redpanda/migrator: avoid O(N²) schema registry fan-out in translate_ids sync

[prakhargarg105]

Recreates #4731 (originally authored by @alextreichler) as a native branch so the Claude review GitHub Action runs. Commit is identical.

Problem

With schema_registry.translate_ids: true, the redpanda_migrator output registers each schema at the destination via franz-go's CreateSchema, which resolves the returned ID through SchemaUsagesByID — a fetch of every subject-version sharing that ID, spawned as one unbounded goroutine per usage.

Identical schema bodies deduplicate to a single destination ID, so syncing N such subjects costs ~N(N+1)/2 destination requests instead of N, and none of those requests are bounded by max_parallel_http_requests. Against a registry with heavily shared schema bodies (e.g. per-environment copies of the same schemas) this produces sustained request bursts that overload single-node registries into connection reset by peer, which fails the sync, which fails the output connect, which restarts the pipeline and replays the whole quadratic sweep — observed in a production migration as 169 consecutive connect failures with zero topics migrated.

A side effect of the fan-out is actively misleading errors: the failing GET names an unrelated subject that merely shares the schema ID (sync subject schema <A> ...: unable to GET ".../subjects/<B>/versions/1"), making healthy subjects look broken.

Fix

In the translate_ids branch of syncSubjectSchema, replace CreateSchema with:

• RegisterSchema — one POST; idempotent (returns the existing ID for an already-registered identical schema), preserving the previous create-or-reuse semantics;
• one LookupSchema — to resolve the destination version for schemaInfo and logging.

Two sequential requests per schema inside the bounded worker pool: O(N) total, concurrency capped by max_parallel_http_requests. franz-go's own RegisterSchema docs recommend exactly this trade. The translate_ids: false path is unchanged.

Measurements

From the included regression test (40 subjects sharing one schema body, max_parallel_http_requests: 2, counting reverse proxy in front of the destination registry):

metric	before	after
usage fan-out reads (GET /subjects/S/versions/V)	826–838 (≈ N(N+1)/2 = 820)	0
peak concurrent in-flight requests	42–79	2 (= configured limit)
schema registrations	40	40

Validation

• New TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout fails on the previous code and passes with the fix.
• All TestIntegrationSchemaRegistryMigrator* integration tests pass except SyncWithReferences, which fails identically on unmodified main (pre-existing, unrelated).
• Unit tests, go vet, gofumpt clean.

Follow-ups (not in this PR)

The same fan-out pattern exists in:

• the migrator's translate_ids: false branch (CreateSchemaWithIDAndVersion fans out internally) — fixable with RegisterSchema(ctx, subject, sch, ss.ID, ss.Version), no lookup needed;
• the internal/impl/confluent/sr wrapper (CreateSchema/CreateSchemaWithIDAndVersion both discard everything but the ID), which the standalone schema_registry output and schema_registry_encode processor inherit;
• separately, the fatal-connect behavior (one failing subject blocks all topic/data migration via the output's Connect) is tracked as its own defect.

🤖 Generated with Claude Code

[squiidz]

The RegisterSchema swap itself is sound — registration semantics are identical to the old path and this removes the SchemaUsagesByID fan-out for the translate_ids: true arm. A few things though, the first one I'd consider blocking since the PR title doesn't hold for the default config (inline comments below).

Two points that fall outside the diff:

The root cause lives in our own wrapper, and two other components still hit it. sr.Client.CreateSchema and CreateSchemaWithIDAndVersion (internal/impl/confluent/sr/client.go:167,181) pay the full SchemaUsagesByID fan-out only to return ss.ID. The schema_registry output (internal/impl/kafka/output_schema_registry.go:478,503 — which has its own translate_ids mode, hit per source subject-version) and the SR encode processor (internal/impl/confluent/processor_schema_registry_encode.go:653, auto-registration on cache miss) call them with the exact same O(N²) behaviour. Switching the two wrapper methods to RegisterSchema is a two-line change that fixes every caller at once — worth doing here or as an immediate follow-up.

max_parallel_http_requests is only enforced as a Sync worker count, not at the HTTP layer, so any other franz-go internal fan-out (e.g. SchemaReferences, the batch endpoints) — or a future library upgrade — silently re-breaks the invariant exactly as CreateSchema did. A semaphore RoundTripper wrapping the client transport with MaxParallelHTTPRequests tokens would make the configured limit unconditionally true, and demote this PR's swap to a request-count optimization rather than the sole enforcement.

[squiidz]

The else branch below (translate_ids: false, the default) still calls CreateSchemaWithIDAndVersion, which performs the exact same SchemaUsagesByID fan-out this PR removes — so the O(N²) behaviour remains for the common configuration.

The fix is even simpler there than here: id and version are inputs (ss.ID, ss.Version), so RegisterSchema alone suffices and info can be built directly as {dstSubject, ss.Version, ss.ID} with no lookup at all. The new fan-out test only covers translate_ids: true, so this path is also untested.

[squiidz]

total is incremented on every request but never read by any assertion or log — either drop it, or better, use it for the missing traffic-happened assertion (see the comment on the assertions below).