From 556807add66b26654378a4c7296581237d21d29e Mon Sep 17 00:00:00 2001 From: alex Date: Thu, 27 Aug 2026 08:31:38 -0700 Subject: [PATCH 1/3] redpanda/migrator: avoid O(N^2) schema registry fan-out in translate_ids sync With translate_ids enabled, each schema was registered 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 cost ~N(N+1)/2 destination requests, none of them bounded by max_parallel_http_requests. Against registries with heavily shared schema bodies this produced request bursts far above the configured concurrency limit (measured: 826 usage reads and peak concurrency 42-79 for 40 subjects with a limit of 2), overloading single-node registries into connection resets. Register with RegisterSchema (one POST, idempotent: returns the existing ID for an already-registered identical schema) plus one LookupSchema to resolve the destination version, making the sync O(N) with concurrency bounded by the worker pool. Measured after: 40 registrations, 0 fan-out reads, peak concurrency exactly at the configured limit. Adds an integration regression test that measures destination traffic through a counting reverse proxy. Co-Authored-By: Claude Fable 5 --- .../migrator/migrator_schema_registry.go | 22 ++- ...schema_registry_fanout_integration_test.go | 164 ++++++++++++++++++ 2 files changed, 183 insertions(+), 3 deletions(-) create mode 100644 internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go diff --git a/internal/impl/redpanda/migrator/migrator_schema_registry.go b/internal/impl/redpanda/migrator/migrator_schema_registry.go index a388bfaf47..4ba5622aaa 100644 --- a/internal/impl/redpanda/migrator/migrator_schema_registry.go +++ b/internal/impl/redpanda/migrator/migrator_schema_registry.go @@ -797,14 +797,30 @@ func (m *schemaRegistryMigrator) syncSubjectSchema(ctx context.Context, ss sr.Su var info schemaInfo t0 := time.Now() if m.conf.TranslateIDs { - // If the schema already exists (and is identical), this returns - // the existing schema - dss, err := m.dst.CreateSchema(ctx, dstSubject, sch) + // Register with a registry-assigned ID. If the schema is already + // registered (and identical), this returns the existing ID without + // creating a new version. + // + // RegisterSchema is used instead of CreateSchema because CreateSchema + // additionally resolves the returned ID via SchemaUsagesByID, which + // fetches every subject-version sharing that ID using one unbounded + // goroutine per usage. Identical schema bodies deduplicate to a single + // ID, so syncing N such subjects costs O(N^2) destination requests, + // none of them bounded by MaxParallelHTTPRequests. + const autoAssign = -1 + id, err := m.dst.RegisterSchema(ctx, dstSubject, sch, autoAssign, autoAssign) if err != nil { m.metrics.IncSchemaCreateErrors() return schemaInfo{}, fmt.Errorf("create schema: %w", err) } + // Single lookup to resolve the destination version for this subject. + dss, err := m.dst.LookupSchema(ctx, dstSubject, sch) + if err != nil { + m.metrics.IncSchemaCreateErrors() + return schemaInfo{}, fmt.Errorf("lookup created schema with id %d: %w", id, err) + } + info = schemaInfoFromSubjectSchema(dss) m.log.Infof("Schema migration: schema created with translated id: subject=%s version=%d id=%d => subject=%s version=%d id=%d", ss.Subject, ss.Version, ss.ID, info.Subject, info.Version, info.ID) diff --git a/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go b/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go new file mode 100644 index 0000000000..628b748d5b --- /dev/null +++ b/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go @@ -0,0 +1,164 @@ +// Copyright 2025 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package migrator_test + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "regexp" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/sr" + + "github.com/redpanda-data/benthos/v4/public/service/integration" + "github.com/redpanda-data/connect/v4/internal/impl/redpanda/migrator" +) + +// countingProxy is a reverse proxy in front of the destination Schema +// Registry that records request counts and peak concurrent in-flight +// requests. +type countingProxy struct { + server *httptest.Server + + total atomic.Int64 // all requests + registers atomic.Int64 // POST /subjects/{s}/versions + idGets atomic.Int64 // GET /schemas/ids/{id}/versions (usage listing) + versionGets atomic.Int64 // GET /subjects/{s}/versions/{v} (usage fan-out) + inFlight atomic.Int64 + maxInFlight atomic.Int64 +} + +var ( + reVersionGet = regexp.MustCompile(`^/subjects/[^/]+/versions/[^/]+$`) + reIDGet = regexp.MustCompile(`^/schemas/ids/\d+/versions$`) + reRegister = regexp.MustCompile(`^/subjects/[^/]+/versions$`) +) + +func newCountingProxy(t *testing.T, targetURL string) *countingProxy { + t.Helper() + + target, err := url.Parse(targetURL) + require.NoError(t, err) + rp := httputil.NewSingleHostReverseProxy(target) + + p := &countingProxy{} + p.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cur := p.inFlight.Add(1) + defer p.inFlight.Add(-1) + for { + maxSeen := p.maxInFlight.Load() + if cur <= maxSeen || p.maxInFlight.CompareAndSwap(maxSeen, cur) { + break + } + } + + p.total.Add(1) + switch { + case r.Method == "GET" && reVersionGet.MatchString(r.URL.Path): + p.versionGets.Add(1) + case r.Method == "GET" && reIDGet.MatchString(r.URL.Path): + p.idGets.Add(1) + case r.Method == "POST" && reRegister.MatchString(r.URL.Path): + p.registers.Add(1) + } + + rp.ServeHTTP(w, r) + })) + t.Cleanup(p.server.Close) + + return p +} + +// TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout guards against +// O(N^2) destination-registry traffic when syncing subjects that share +// identical schema bodies with translate_ids enabled. +// +// Identical schema bodies deduplicate to a single destination schema ID. +// Registering via franz-go's CreateSchema resolves the returned ID through +// SchemaUsagesByID, which fetches every subject-version sharing that ID using +// one unbounded goroutine per usage: N such subjects cost ~N(N+1)/2 requests, +// none of them bounded by max_parallel_http_requests. The sync must instead +// stay O(N) and respect the configured concurrency limit. +func TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout(t *testing.T) { + integration.CheckSkip(t) + + // Number of source subjects sharing one identical schema body. Large + // enough to make quadratic growth unambiguous while keeping runtime low. + const numSubjects = 40 + + t.Log("Given: source and destination Redpanda clusters with Schema Registry") + srcCluster, dstCluster := startRedpandaSourceAndDestination(t) + + srcSR, err := sr.NewClient(sr.URLs(srcCluster.SchemaRegistryURL)) + require.NoError(t, err) + + t.Log("And: a counting reverse proxy in front of the destination Schema Registry") + proxy := newCountingProxy(t, dstCluster.SchemaRegistryURL) + dstSR, err := sr.NewClient(sr.URLs(proxy.server.URL)) + require.NoError(t, err) + + t.Logf("And: %d source subjects sharing one identical schema body", numSubjects) + const sharedSchema = `{"type":"record","name":"Shared","fields":[{"name":"a","type":"int"}]}` + ctx := t.Context() + for i := range numSubjects { + _, err := srcSR.CreateSchema(ctx, fmt.Sprintf("shared-%03d-value", i), sr.Schema{Schema: sharedSchema}) + require.NoError(t, err) + } + + t.Log("When: the schema migrator syncs with translate_ids enabled") + conf := migrator.SchemaRegistryMigratorConfig{ + Enabled: true, + Versions: migrator.VersionsAll, + TranslateIDs: true, + } + // NB: the testing constructor sets MaxParallelHTTPRequests to 2. + m := migrator.NewSchemaRegistryMigratorForTesting(t, conf, srcSR, dstSR) + + syncCtx, cancel := context.WithTimeout(ctx, 3*time.Minute) + defer cancel() + require.NoError(t, m.Sync(syncCtx)) + + registers := proxy.registers.Load() + idGets := proxy.idGets.Load() + versionGets := proxy.versionGets.Load() + maxInFlight := proxy.maxInFlight.Load() + + t.Logf("Destination registry traffic for %d identical-body subjects:", numSubjects) + t.Logf(" schema registrations (POST): %d", registers) + t.Logf(" usage listings (GET /schemas/ids/N/versions): %d", idGets) + t.Logf(" usage fan-out (GET /subjects/S/versions/V): %d (O(N) expectation: <=%d, O(N^2) worst case: %d)", + versionGets, numSubjects, numSubjects*(numSubjects+1)/2) + t.Logf(" peak concurrent in-flight requests: %d (max_parallel_http_requests: 2)", maxInFlight) + + // Syncing N subjects must cost O(N) destination requests. 2*N leaves room + // for one extra lookup per subject. + assert.LessOrEqual(t, versionGets, int64(2*numSubjects), + "O(N^2) subject-version GETs against the destination registry - "+ + "schema registration must not fan out to every subject sharing the "+ + "destination schema ID") + + // Concurrency against the destination registry must respect + // max_parallel_http_requests (2 here). + assert.LessOrEqual(t, maxInFlight, int64(2), + "destination request concurrency exceeds max_parallel_http_requests") +} From f32a24832cc65575d725b9569c9e3b29718a117b Mon Sep 17 00:00:00 2001 From: alex Date: Thu, 27 Aug 2026 11:46:26 -0700 Subject: [PATCH 2/3] redpanda/migrator: address review - drop version lookup, document test Drop the LookupSchema call after RegisterSchema: the destination version it resolved only fed a log field (schemaInfo's sole functional consumer is the ID), and a transient lookup failure would have aborted an otherwise-successful registration. The translated-ID sync now costs exactly one request per schema. Add the new fan-out integration test to the TESTING.md catalog. Co-Authored-By: Claude Fable 5 --- internal/impl/redpanda/migrator/TESTING.md | 13 +++++++++++++ .../migrator/migrator_schema_registry.go | 16 ++++++---------- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/internal/impl/redpanda/migrator/TESTING.md b/internal/impl/redpanda/migrator/TESTING.md index 75c8679d3b..3f68c9c970 100644 --- a/internal/impl/redpanda/migrator/TESTING.md +++ b/internal/impl/redpanda/migrator/TESTING.md @@ -195,6 +195,19 @@ Tests migration of compatibility mode settings. - Validates compatibility mode is preserved - Tests various compatibility levels (BACKWARD, FORWARD, FULL, etc.) +## Schema Registry Fan-out Test (`migrator_schema_registry_fanout_integration_test.go`) + +### `TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout` + +Guards against O(N^2) destination-registry traffic when syncing subjects that share identical schema bodies with `translate_ids` enabled. +- Creates source and destination clusters with Schema Registry +- Places a counting reverse proxy in front of the destination Schema Registry, recording request counts by endpoint and peak concurrent in-flight requests +- Registers 40 source subjects sharing one identical schema body (which deduplicate to a single destination schema ID) +- Syncs with `translate_ids: true` and `max_parallel_http_requests: 2` +- Validates: + - Destination subject-version reads stay O(N) (no per-registration fan-out to every subject sharing the destination schema ID) + - Peak destination request concurrency respects `max_parallel_http_requests` + ## Topic Migration Tests (`migrator_topic_integration_test.go`) ### `TestIntegrationTopicMigratorSyncConfig` diff --git a/internal/impl/redpanda/migrator/migrator_schema_registry.go b/internal/impl/redpanda/migrator/migrator_schema_registry.go index 4ba5622aaa..601038109d 100644 --- a/internal/impl/redpanda/migrator/migrator_schema_registry.go +++ b/internal/impl/redpanda/migrator/migrator_schema_registry.go @@ -814,16 +814,12 @@ func (m *schemaRegistryMigrator) syncSubjectSchema(ctx context.Context, ss sr.Su return schemaInfo{}, fmt.Errorf("create schema: %w", err) } - // Single lookup to resolve the destination version for this subject. - dss, err := m.dst.LookupSchema(ctx, dstSubject, sch) - if err != nil { - m.metrics.IncSchemaCreateErrors() - return schemaInfo{}, fmt.Errorf("lookup created schema with id %d: %w", id, err) - } - - info = schemaInfoFromSubjectSchema(dss) - m.log.Infof("Schema migration: schema created with translated id: subject=%s version=%d id=%d => subject=%s version=%d id=%d", - ss.Subject, ss.Version, ss.ID, info.Subject, info.Version, info.ID) + // The destination version is left unset: the registration response + // carries only the ID, which is also the only field of schemaInfo + // with a functional consumer. + info = schemaInfo{Subject: dstSubject, ID: id} + m.log.Infof("Schema migration: schema created with translated id: subject=%s version=%d id=%d => subject=%s id=%d", + ss.Subject, ss.Version, ss.ID, info.Subject, info.ID) } else { dss, err := m.dst.CreateSchemaWithIDAndVersion(ctx, dstSubject, sch, ss.ID, ss.Version) if err != nil { From 575e7388ba15539d87bb95a698b9f3b71f1104cb Mon Sep 17 00:00:00 2001 From: alex Date: Thu, 27 Aug 2026 12:11:58 -0700 Subject: [PATCH 3/3] redpanda/migrator: address review - fix fan-out in fixed-ID path, harden test Extend the RegisterSchema swap to the translate_ids: false branch (the default configuration), where CreateSchemaWithIDAndVersion performed the same unbounded SchemaUsagesByID fan-out; the ID and version are inputs there, so no lookup is needed at all. The redpanda#26331 fallback is preserved unchanged. Harden the fan-out regression test per review: cover both ID-translation modes (table-driven, IMPORT destination for fixed IDs), seed the source via RegisterSchema so setup does not itself fan out, guard against a vacuous pass by requiring one registration per subject, tighten the fan-out bounds to exactly zero usage-endpoint requests, use the package's standard sync timeout, and fix the new file's copyright year. Remove the now-unused schemaInfoFromSubjectSchema. Co-Authored-By: Claude Fable 5 --- internal/impl/redpanda/migrator/TESTING.md | 9 +- .../migrator/migrator_schema_registry.go | 19 +-- ...schema_registry_fanout_integration_test.go | 148 +++++++++++------- 3 files changed, 100 insertions(+), 76 deletions(-) diff --git a/internal/impl/redpanda/migrator/TESTING.md b/internal/impl/redpanda/migrator/TESTING.md index 3f68c9c970..4922691286 100644 --- a/internal/impl/redpanda/migrator/TESTING.md +++ b/internal/impl/redpanda/migrator/TESTING.md @@ -199,13 +199,14 @@ Tests migration of compatibility mode settings. ### `TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout` -Guards against O(N^2) destination-registry traffic when syncing subjects that share identical schema bodies with `translate_ids` enabled. -- Creates source and destination clusters with Schema Registry +Guards against O(N^2) destination-registry traffic when syncing subjects that share identical schema bodies, in both ID-translation modes. +- Creates source and destination clusters with Schema Registry (per subtest: `translate_ids: true` with READWRITE destination, `translate_ids: false` with IMPORT destination) - Places a counting reverse proxy in front of the destination Schema Registry, recording request counts by endpoint and peak concurrent in-flight requests - Registers 40 source subjects sharing one identical schema body (which deduplicate to a single destination schema ID) -- Syncs with `translate_ids: true` and `max_parallel_http_requests: 2` +- Syncs with `max_parallel_http_requests: 2` - Validates: - - Destination subject-version reads stay O(N) (no per-registration fan-out to every subject sharing the destination schema ID) + - At least one destination registration per subject occurred (guards against a vacuous pass) + - Zero requests to the schema-usage endpoints (no per-registration fan-out to the subject-versions sharing the destination schema ID) - Peak destination request concurrency respects `max_parallel_http_requests` ## Topic Migration Tests (`migrator_topic_integration_test.go`) diff --git a/internal/impl/redpanda/migrator/migrator_schema_registry.go b/internal/impl/redpanda/migrator/migrator_schema_registry.go index 601038109d..4849f2b337 100644 --- a/internal/impl/redpanda/migrator/migrator_schema_registry.go +++ b/internal/impl/redpanda/migrator/migrator_schema_registry.go @@ -367,14 +367,6 @@ type schemaInfo struct { ID int } -func schemaInfoFromSubjectSchema(ss sr.SubjectSchema) schemaInfo { - return schemaInfo{ - Subject: ss.Subject, - Version: ss.Version, - ID: ss.ID, - } -} - // schemaRegistryMigrator coordinates migration between a source and destination // Schema Registry. // @@ -821,7 +813,11 @@ func (m *schemaRegistryMigrator) syncSubjectSchema(ctx context.Context, ss sr.Su m.log.Infof("Schema migration: schema created with translated id: subject=%s version=%d id=%d => subject=%s id=%d", ss.Subject, ss.Version, ss.ID, info.Subject, info.ID) } else { - dss, err := m.dst.CreateSchemaWithIDAndVersion(ctx, dstSubject, sch, ss.ID, ss.Version) + // RegisterSchema instead of CreateSchemaWithIDAndVersion for the same + // reason as above: the latter resolves the registered ID through the + // unbounded SchemaUsagesByID fan-out, and the ID and version are + // already known here. + id, err := m.dst.RegisterSchema(ctx, dstSubject, sch, ss.ID, ss.Version) if err != nil { const conflictPattern = `Schema already registered with id \d+ instead of input id \d+` if ok, _ := regexp.MatchString(conflictPattern, err.Error()); ok { @@ -844,11 +840,10 @@ func (m *schemaRegistryMigrator) syncSubjectSchema(ctx context.Context, ss sr.Su m.log.Warnf("Schema migration: schema subject=%s version=%d id=%d could not be created (server error: %s) - using existing schema with the same ID, if this is not the desired behavior, try enabling translate-ids", ss.Subject, ss.Version, ss.ID, err.Error()) - dss = ss - dss.Subject = dstSubject + id = ss.ID } - info = schemaInfoFromSubjectSchema(dss) + info = schemaInfo{Subject: dstSubject, Version: ss.Version, ID: id} m.log.Infof("Schema migration: schema created with fixed id: subject=%s version=%d id=%d", info.Subject, info.Version, info.ID) } diff --git a/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go b/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go index 628b748d5b..11e9c59b4c 100644 --- a/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go +++ b/internal/impl/redpanda/migrator/migrator_schema_registry_fanout_integration_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Redpanda Data, Inc. +// Copyright 2026 Redpanda Data, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -24,7 +24,6 @@ import ( "regexp" "sync/atomic" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -91,74 +90,103 @@ func newCountingProxy(t *testing.T, targetURL string) *countingProxy { // TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout guards against // O(N^2) destination-registry traffic when syncing subjects that share -// identical schema bodies with translate_ids enabled. +// identical schema bodies, in both ID-translation modes. // // Identical schema bodies deduplicate to a single destination schema ID. -// Registering via franz-go's CreateSchema resolves the returned ID through -// SchemaUsagesByID, which fetches every subject-version sharing that ID using -// one unbounded goroutine per usage: N such subjects cost ~N(N+1)/2 requests, -// none of them bounded by max_parallel_http_requests. The sync must instead -// stay O(N) and respect the configured concurrency limit. +// Registering via franz-go's CreateSchema/CreateSchemaWithIDAndVersion +// resolves the returned ID through SchemaUsagesByID, which fetches every +// subject-version sharing that ID using one unbounded goroutine per usage: N +// such subjects cost ~N(N+1)/2 requests, none of them bounded by +// max_parallel_http_requests. The sync must instead cost one registration per +// subject with no usage fan-out, and respect the configured concurrency +// limit. func TestIntegrationSchemaRegistryMigratorSyncSharedSchemaFanout(t *testing.T) { integration.CheckSkip(t) // Number of source subjects sharing one identical schema body. Large // enough to make quadratic growth unambiguous while keeping runtime low. const numSubjects = 40 - - t.Log("Given: source and destination Redpanda clusters with Schema Registry") - srcCluster, dstCluster := startRedpandaSourceAndDestination(t) - - srcSR, err := sr.NewClient(sr.URLs(srcCluster.SchemaRegistryURL)) - require.NoError(t, err) - - t.Log("And: a counting reverse proxy in front of the destination Schema Registry") - proxy := newCountingProxy(t, dstCluster.SchemaRegistryURL) - dstSR, err := sr.NewClient(sr.URLs(proxy.server.URL)) - require.NoError(t, err) - - t.Logf("And: %d source subjects sharing one identical schema body", numSubjects) const sharedSchema = `{"type":"record","name":"Shared","fields":[{"name":"a","type":"int"}]}` - ctx := t.Context() - for i := range numSubjects { - _, err := srcSR.CreateSchema(ctx, fmt.Sprintf("shared-%03d-value", i), sr.Schema{Schema: sharedSchema}) - require.NoError(t, err) + + tests := []struct { + name string + translate bool + mode sr.Mode + }{ + {name: "translate_ids=true", translate: true, mode: sr.ModeReadWrite}, + {name: "translate_ids=false", translate: false, mode: sr.ModeImport}, } - t.Log("When: the schema migrator syncs with translate_ids enabled") - conf := migrator.SchemaRegistryMigratorConfig{ - Enabled: true, - Versions: migrator.VersionsAll, - TranslateIDs: true, + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Log("Given: source and destination Redpanda clusters with Schema Registry") + srcCluster, dstCluster := startRedpandaSourceAndDestination(t) + + srcSR, err := sr.NewClient(sr.URLs(srcCluster.SchemaRegistryURL)) + require.NoError(t, err) + + t.Log("And: a counting reverse proxy in front of the destination Schema Registry") + proxy := newCountingProxy(t, dstCluster.SchemaRegistryURL) + dstSR, err := sr.NewClient(sr.URLs(proxy.server.URL)) + require.NoError(t, err) + + t.Logf("And: destination is set to %s mode", tc.mode) + modeRes := dstSR.SetMode(t.Context(), tc.mode) + require.NoError(t, modeRes[0].Err) + + t.Logf("And: %d source subjects sharing one identical schema body", numSubjects) + const autoAssign = -1 + for i := range numSubjects { + // RegisterSchema: CreateSchema would perform the same usage + // fan-out this test guards against, against the source. + _, err := srcSR.RegisterSchema(t.Context(), + fmt.Sprintf("shared-%03d-value", i), + sr.Schema{Schema: sharedSchema}, autoAssign, autoAssign) + require.NoError(t, err) + } + + t.Logf("When: the schema migrator syncs with translate_ids=%v", tc.translate) + conf := migrator.SchemaRegistryMigratorConfig{ + Enabled: true, + Versions: migrator.VersionsAll, + TranslateIDs: tc.translate, + } + // NB: the testing constructor sets MaxParallelHTTPRequests to 2. + m := migrator.NewSchemaRegistryMigratorForTesting(t, conf, srcSR, dstSR) + + ctx, cancel := context.WithTimeout(t.Context(), redpandaTestWaitTimeout) + defer cancel() + require.NoError(t, m.Sync(ctx)) + + registers := proxy.registers.Load() + idGets := proxy.idGets.Load() + versionGets := proxy.versionGets.Load() + maxInFlight := proxy.maxInFlight.Load() + + t.Logf("Destination registry traffic for %d identical-body subjects:", numSubjects) + t.Logf(" total requests: %d", proxy.total.Load()) + t.Logf(" schema registrations (POST): %d", registers) + t.Logf(" usage listings (GET /schemas/ids/N/versions): %d", idGets) + t.Logf(" usage fan-out (GET /subjects/S/versions/V): %d (O(N^2) worst case: %d)", + versionGets, numSubjects*(numSubjects+1)/2) + t.Logf(" peak concurrent in-flight requests: %d (max_parallel_http_requests: 2)", maxInFlight) + + // Guard against a vacuous pass: the sync must actually have + // registered every subject at the destination. + assert.GreaterOrEqual(t, registers, int64(numSubjects), + "expected at least one registration per subject") + + // Registration must not resolve its result through the usage + // endpoints: any hit is the start of the O(N^2) fan-out. + assert.Zero(t, versionGets, + "schema registration must not fan out to the subject-versions sharing the destination schema ID") + assert.Zero(t, idGets, + "schema registration must not list usages of the destination schema ID") + + // Concurrency against the destination registry must respect + // max_parallel_http_requests (2 here). + assert.LessOrEqual(t, maxInFlight, int64(2), + "destination request concurrency exceeds max_parallel_http_requests") + }) } - // NB: the testing constructor sets MaxParallelHTTPRequests to 2. - m := migrator.NewSchemaRegistryMigratorForTesting(t, conf, srcSR, dstSR) - - syncCtx, cancel := context.WithTimeout(ctx, 3*time.Minute) - defer cancel() - require.NoError(t, m.Sync(syncCtx)) - - registers := proxy.registers.Load() - idGets := proxy.idGets.Load() - versionGets := proxy.versionGets.Load() - maxInFlight := proxy.maxInFlight.Load() - - t.Logf("Destination registry traffic for %d identical-body subjects:", numSubjects) - t.Logf(" schema registrations (POST): %d", registers) - t.Logf(" usage listings (GET /schemas/ids/N/versions): %d", idGets) - t.Logf(" usage fan-out (GET /subjects/S/versions/V): %d (O(N) expectation: <=%d, O(N^2) worst case: %d)", - versionGets, numSubjects, numSubjects*(numSubjects+1)/2) - t.Logf(" peak concurrent in-flight requests: %d (max_parallel_http_requests: 2)", maxInFlight) - - // Syncing N subjects must cost O(N) destination requests. 2*N leaves room - // for one extra lookup per subject. - assert.LessOrEqual(t, versionGets, int64(2*numSubjects), - "O(N^2) subject-version GETs against the destination registry - "+ - "schema registration must not fan out to every subject sharing the "+ - "destination schema ID") - - // Concurrency against the destination registry must respect - // max_parallel_http_requests (2 here). - assert.LessOrEqual(t, maxInFlight, int64(2), - "destination request concurrency exceeds max_parallel_http_requests") }