redpanda/migrator: set IMPORT mode per subject instead of per version for serverless SR by mmatczuk · Pull Request #4090 · redpanda-data/connect

mmatczuk · 2026-03-11T12:29:32Z

Introduce importModeManager to batch per-subject IMPORT mode transitions in serverless schema registries. Previously ensureSubjectImportMode was called inside syncSubjectSchema for every version, toggling IMPORT mode on and off repeatedly. The new manager sets IMPORT mode once before the first version of a subject is written and restores the original mode after all versions are migrated. Close acts as a safety net, retrying restore for any subjects left over on error paths.

internal/impl/redpanda/migrator/migrator_schema_registry.go

claude · 2026-03-11T13:47:24Z

internal/impl/redpanda/migrator/migrator_schema_registry.go

+// its previous mode once all its versions have been migrated.
+//
+// When the destination global mode is already IMPORT, or when the migrator is
+// not in serverless mode, all operations are no-ops.


Bug: resolveSubject(srcSubject, 0) passes a hardcoded version 0, but resolveSubject exposes schema_registry_version as interpolation metadata for the NameResolver. If the name resolver references that metadata, version 0 produces an incorrect destination subject, and IMPORT mode gets set on the wrong subject.

The old code avoided this because ensureSubjectImportMode was called from syncSubjectSchema with the already-resolved dstSubject (which used ss.Version). Now that TrySetImportMode resolves the subject itself, it needs the actual version — or, preferably, accept the already-resolved dstSubject to stay consistent with the previous approach and avoid duplicating resolution logic.

internal/impl/redpanda/migrator/migrator_schema_registry.go

claude · 2026-03-11T15:20:52Z

internal/impl/redpanda/migrator/migrator_schema_registry.go

+// TrySetImportMode sets the subject to IMPORT mode if not already done.
+// No-op when inactive or when the subject was already switched.
+func (c *importModeManager) TrySetImportMode(ctx context.Context, src schemaSubjectVersion) error {
+	if !c.active {
+		return nil
+	}
+
+	dstSubject, err := c.resolveSubject(src.Subject, src.Version)
+	if err != nil {
+		return err
+	}
+
+	// Fast path: subject already tracked.
+	c.mu.RLock()
+	_, ok := c.prevMode[src]
+	c.mu.RUnlock()
+	if ok {
+		return nil
+	}
+
+	// Slow path: hold exclusive lock across the entire check-fetch-set
+	// sequence to prevent concurrent goroutines from racing on the same
+	// subject and clobbering the original mode.
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if _, ok := c.prevMode[src]; ok {
+		return nil
+	}
+
+	mode, err := srSubjectMode(ctx, c.dst, dstSubject)
 	if err != nil {
 		if strings.Contains(err.Error(), "does not have subject-level mode configured") {
 			mode = noMode
 		} else {
-			return noop, err
+			return err
 		}
 	} else if mode == sr.ModeImport {
-		return noop, nil
+		c.prevMode[src] = subjectMode{Subject: dstSubject, Mode: mode}
+		return nil


Bug: prevMode is keyed by source schemaSubjectVersion (subject+version), but the IMPORT mode operations target the destination subject. When VersionsAll is configured, the DFS processes multiple versions of the same subject sequentially via TrySetImportMode. The first version (e.g. C-v1) correctly records the original destination mode (READWRITE), sets IMPORT, and stores prevMode[C-v1] = {dst, READWRITE}. The second version (C-v2) then fetches the destination mode (now IMPORT), hits the else if mode == sr.ModeImport branch, and stores prevMode[C-v2] = {dst, IMPORT}.

During restore (in Close()), if C-v1 is restored first (→ READWRITE) and then C-v2 is restored (→ IMPORT), the subject is left in IMPORT mode. Since Go map iteration order is non-deterministic, this produces the wrong final state ~50% of the time.

The fix: key prevMode by destination subject instead of source subject+version, so only the first call per destination subject records the original mode and subsequent versions are no-ops.

claude · 2026-03-11T16:31:20Z

Commits
LGTM

Review
This PR introduces importModeManager to batch per-subject IMPORT mode transitions in serverless schema registries, replacing the previous per-version toggling. The new manager sets IMPORT mode once before the first version of a subject is written and restores the original mode after all versions are migrated, with Close() as a safety net for error paths. The concurrency model is sound — Sync always uses VersionsLatest so each subject gets exactly one channel item with Last: true, avoiding races between Restore and TrySetImportMode. Tests cover multi-version subjects with shared dependencies, single subjects, and pre-existing IMPORT mode.

LGTM

claude · 2026-03-11T16:35:46Z

internal/impl/redpanda/migrator/migrator_schema_registry.go

 						err = fmt.Errorf("get schema for subject %q version %d: %w", s, v, err)
 					}
-					if !yield(schema, err) {
+					if !yield(subjectSchema{SubjectSchema: schema, Last: i == len(subs)-1}, err) {


Bug: Race condition — Last is i == len(subs)-1, meaning it's true for all versions of the last subject (by index), not the last version of each subject. This creates two problems:

Race with concurrent workers: In Sync, items are distributed to MaxParallelHTTPRequests workers via a channel. Multiple versions of the last subject can be processed concurrently by different workers. The first worker to finish any version of the last subject calls Restore (line 709), reverting IMPORT mode while other workers may still be syncing other versions of that same subject — causing those in-flight syncs to fail.

Non-last subjects never get Restore called by workers: For all subjects except the last one, Last is always false, so their IMPORT mode is only restored during Close() after the entire sync completes. This holds IMPORT mode longer than necessary.

The Last flag needs to track whether a given item is the last version for its subject, and Restore should only be called after all versions of that subject have been fully processed — which requires coordination across workers (e.g., a per-subject counter or processing all versions of a subject within a single worker).

…n for serverless SR Introduce importModeManager to batch per-subject IMPORT mode transitions in serverless schema registries. Previously ensureSubjectImportMode was called inside syncSubjectSchema for every version, toggling IMPORT mode on and off repeatedly. The new manager sets IMPORT mode once before the first version of a subject is written and restores the original mode after all versions are migrated. Close acts as a safety net, retrying restore for any subjects left over on error paths.

claude · 2026-03-11T16:35:56Z

Commits
LGTM

Review
The PR refactors per-version IMPORT mode toggling into a centralized importModeManager with double-checked locking, deferred cleanup, and retry logic. The new integration tests cover multi-version, shared-dependency, and pre-set IMPORT scenarios well.

Bug: Race condition with Last flag in VersionsAll mode — In listSubjectSchemas, Last is set based on whether the subject is the last in the list (i == len(subs)-1), not whether the version is the last for its subject. Since Sync fans items out to concurrent workers, a worker finishing early can call Restore (reverting IMPORT mode) while other workers are still syncing other versions of the same subject. For non-last subjects, Restore is never called by workers at all — they rely entirely on Close(). See inline comment.

claude · 2026-03-11T16:40:47Z

internal/impl/redpanda/migrator/migrator_schema_registry.go

 						err = fmt.Errorf("get schema for subject %q version %d: %w", s, v, err)
 					}
-					if !yield(schema, err) {
+					if !yield(subjectSchema{SubjectSchema: schema, Last: i == len(subs)-1}, err) {


Bug: Last: i == len(subs)-1 marks all versions of the last subject as Last: true, and no versions of any earlier subject. The intended semantic (based on how Sync uses Last to trigger Restore) is "last version of each subject", which should be checking the inner loop index against len(vers)-1, not the outer loop index against len(subs)-1.

Currently this is masked because Sync always calls listSubjectSchemas with VersionsLatest (where Last: true is unconditional), so the VersionsAll path's Last value is never consumed by Sync. However, ListSubjectSchemas does call this with VersionsAll and silently carries the wrong Last value (only ignored because it discards it via ss.SubjectSchema). If Sync is ever changed to use VersionsAll, this would cause:

Intermediate subjects never eagerly restored (relying entirely on Close() safety net)

A race condition for the last subject — multiple workers processing different versions could call Restore while another worker is still syncing, removing IMPORT mode mid-write

claude · 2026-03-11T16:40:57Z

Commits
LGTM

Review
This PR introduces importModeManager to batch per-subject IMPORT mode transitions for serverless schema registries, replacing the previous per-version toggling. The overall design is sound — the manager correctly deduplicates mode transitions, handles concurrent access with a read/write mutex, and provides a Close() safety net for error paths.

Last field logic incorrect for VersionsAll: In listSubjectSchemas, the Last flag in the VersionsAll case uses the outer subject index (i == len(subs)-1) instead of the inner version index, marking all versions of only the last subject as Last. Currently masked because Sync always uses VersionsLatest, but would cause race conditions and missed eager restores if Sync ever switches to VersionsAll.