feat: implement reserved fields validation (validateFields config)

Add Confluent-compatible reserved fields validation controlled by the
validateFields config option. When enabled, two rules are enforced:
1. Reserved fields (listed in confluent:reserved metadata) must not
   conflict with actual top-level schema fields
2. Reserved fields from previous versions must not be removed

Changes:
- Add HasTopLevelField() to ParsedSchema interface (Avro, Protobuf, JSON Schema)
- Add ValidateFields config to storage, API types, and all 4 backends
- Add validateReservedFields() and isValidateFieldsEnabled() to registry
- Add DB migrations for validate_fields column (Postgres, MySQL, Cassandra)
- Add 12 BDD scenarios covering all schema types and config overrides

Author: millerjp

internal/api/handlers/handlers.go

@@ -575,6 +575,7 @@ func (h *Handler) GetConfig(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusOK, types.ConfigResponse{
 			CompatibilityLevel: config.CompatibilityLevel,
 			Normalize:          config.Normalize,
+			ValidateFields:     config.ValidateFields,
 			Alias:              config.Alias,
 			CompatibilityGroup: config.CompatibilityGroup,
 			DefaultMetadata:    config.DefaultMetadata,
@@ -594,6 +595,7 @@ func (h *Handler) GetConfig(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, types.ConfigResponse{
 		CompatibilityLevel: config.CompatibilityLevel,
 		Normalize:          config.Normalize,
+		ValidateFields:     config.ValidateFields,
 		Alias:              config.Alias,
 		CompatibilityGroup: config.CompatibilityGroup,
 		DefaultMetadata:    config.DefaultMetadata,
@@ -627,6 +629,7 @@ func (h *Handler) SetConfig(w http.ResponseWriter, r *http.Request) {
 	configOpts := registry.SetConfigOpts{
 		Alias:              req.Alias,
 		CompatibilityGroup: req.CompatibilityGroup,
+		ValidateFields:     req.ValidateFields,
 		DefaultMetadata:    req.DefaultMetadata,
 		OverrideMetadata:   req.OverrideMetadata,
 		DefaultRuleSet:     req.DefaultRuleSet,
@@ -644,6 +647,7 @@ func (h *Handler) SetConfig(w http.ResponseWriter, r *http.Request) {
 	resp := types.ConfigRequest{
 		Compatibility:      strings.ToUpper(req.Compatibility),
 		Normalize:          req.Normalize,
+		ValidateFields:     req.ValidateFields,
 		Alias:              req.Alias,
 		CompatibilityGroup: req.CompatibilityGroup,
 		DefaultMetadata:    req.DefaultMetadata,

internal/api/types/types.go

@@ -70,6 +70,7 @@ type LookupSchemaResponse struct {
 type ConfigResponse struct {
 	CompatibilityLevel string            `json:"compatibilityLevel"`
 	Normalize          *bool             `json:"normalize,omitempty"`
+	ValidateFields     *bool             `json:"validateFields,omitempty"`
 	Alias              string            `json:"alias,omitempty"`
 	CompatibilityGroup string            `json:"compatibilityGroup,omitempty"`
 	DefaultMetadata    *storage.Metadata `json:"defaultMetadata,omitempty"`
@@ -82,6 +83,7 @@ type ConfigResponse struct {
 type ConfigRequest struct {
 	Compatibility      string            `json:"compatibility"`
 	Normalize          *bool             `json:"normalize,omitempty"`
+	ValidateFields     *bool             `json:"validateFields,omitempty"`
 	Alias              string            `json:"alias,omitempty"`
 	CompatibilityGroup string            `json:"compatibilityGroup,omitempty"`
 	DefaultMetadata    *storage.Metadata `json:"defaultMetadata,omitempty"`

internal/registry/registry.go

@@ -135,6 +135,13 @@ func (r *Registry) RegisterSchema(ctx context.Context, subject string, schemaStr
 		}
 	}
 
+	// Validate reserved fields if enabled
+	if r.isValidateFieldsEnabled(ctx, subject) {
+		if msgs := r.validateReservedFields(ctx, subject, parsed, opt.Metadata); len(msgs) > 0 {
+			return nil, fmt.Errorf("%w: %s", ErrIncompatibleSchema, strings.Join(msgs, "; "))
+		}
+	}
+
 	// Check confluent:version (compare-and-set) if present in metadata
 	if err := r.checkConfluentVersion(ctx, subject, opt.Metadata); err != nil {
 		return nil, err
@@ -627,6 +634,7 @@ func (r *Registry) GetConfigFull(ctx context.Context, subject string) (*storage.
 type SetConfigOpts struct {
 	Alias              string
 	CompatibilityGroup string
+	ValidateFields     *bool
 	DefaultMetadata    *storage.Metadata
 	OverrideMetadata   *storage.Metadata
 	DefaultRuleSet     *storage.RuleSet
@@ -649,6 +657,7 @@ func (r *Registry) SetConfig(ctx context.Context, subject string, level string,
 		opt := opts[0]
 		config.Alias = opt.Alias
 		config.CompatibilityGroup = opt.CompatibilityGroup
+		config.ValidateFields = opt.ValidateFields
 		config.DefaultMetadata = opt.DefaultMetadata
 		config.OverrideMetadata = opt.OverrideMetadata
 		config.DefaultRuleSet = opt.DefaultRuleSet
@@ -1060,6 +1069,77 @@ func (r *Registry) filterByCompatibilityGroup(ctx context.Context, subject strin
 	return filtered
 }
 
+// isValidateFieldsEnabled checks if reserved field validation is enabled for a subject.
+func (r *Registry) isValidateFieldsEnabled(ctx context.Context, subject string) bool {
+	// Check subject config first
+	if subject != "" {
+		config, err := r.storage.GetConfig(ctx, subject)
+		if err == nil && config != nil && config.ValidateFields != nil {
+			return *config.ValidateFields
+		}
+	}
+	// Fall back to global config
+	config, err := r.storage.GetGlobalConfig(ctx)
+	if err == nil && config != nil && config.ValidateFields != nil {
+		return *config.ValidateFields
+	}
+	return false
+}
+
+// getReservedFields extracts reserved field names from the "confluent:reserved"
+// metadata property. Returns an empty set if not present.
+func getReservedFields(metadata *storage.Metadata) map[string]bool {
+	if metadata == nil || metadata.Properties == nil {
+		return nil
+	}
+	val, ok := metadata.Properties["confluent:reserved"]
+	if !ok || val == "" {
+		return nil
+	}
+	fields := make(map[string]bool)
+	for _, f := range strings.Split(val, ",") {
+		f = strings.TrimSpace(f)
+		if f != "" {
+			fields[f] = true
+		}
+	}
+	if len(fields) == 0 {
+		return nil
+	}
+	return fields
+}
+
+// validateReservedFields checks two invariants when validateFields is enabled:
+// 1. Reserved fields listed in metadata must not exist as top-level schema fields.
+// 2. Reserved fields from the previous version must not be removed.
+func (r *Registry) validateReservedFields(ctx context.Context, subject string, parsed schema.ParsedSchema, metadata *storage.Metadata) []string {
+	var msgs []string
+
+	reservedFields := getReservedFields(metadata)
+
+	// Rule 2: Check that reserved fields from previous version are not removed
+	latest, err := r.storage.GetLatestSchema(ctx, subject)
+	if err == nil && latest != nil {
+		prevReserved := getReservedFields(latest.Metadata)
+		for field := range prevReserved {
+			if !reservedFields[field] {
+				msgs = append(msgs, fmt.Sprintf(
+					"The new schema has reserved field %s removed from its metadata which is present in the old schema's metadata.", field))
+			}
+		}
+	}
+
+	// Rule 1: Reserved fields must not conflict with actual schema fields
+	for field := range reservedFields {
+		if parsed.HasTopLevelField(field) {
+			msgs = append(msgs, fmt.Sprintf(
+				"The new schema has field that conflicts with the reserved field %s.", field))
+		}
+	}
+
+	return msgs
+}
+
 // resolveReferences looks up the schema content for each reference from storage.
 func (r *Registry) resolveReferences(ctx context.Context, refs []storage.Reference) ([]storage.Reference, error) {
 	if len(refs) == 0 {

internal/schema/avro/parser.go

@@ -104,6 +104,19 @@ func (s *ParsedSchema) Normalize() schema.ParsedSchema {
 	}
 }
 
+// HasTopLevelField reports whether the Avro record schema contains a field
+// with the given name. Returns false for non-record schemas.
+func (s *ParsedSchema) HasTopLevelField(field string) bool {
+	if rs, ok := s.rawSchema.(*avro.RecordSchema); ok {
+		for _, f := range rs.Fields() {
+			if f.Name() == field {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // FormattedString returns the schema in the requested format.
 // Supported formats: "resolved" (inlines all references), "default" (canonical).
 func (s *ParsedSchema) FormattedString(format string) string {

internal/schema/jsonschema/parser.go

@@ -115,6 +115,17 @@ func (p *ParsedJSONSchema) Normalize() schema.ParsedSchema {
 	}
 }
 
+// HasTopLevelField reports whether the JSON Schema "properties" object
+// contains a key with the given name.
+func (p *ParsedJSONSchema) HasTopLevelField(field string) bool {
+	props, ok := p.schemaMap["properties"].(map[string]interface{})
+	if !ok {
+		return false
+	}
+	_, exists := props[field]
+	return exists
+}
+
 // FormattedString returns the schema in the requested format.
 // JSON Schema does not support special format values; always returns canonical string.
 func (p *ParsedJSONSchema) FormattedString(format string) string {

internal/schema/protobuf/parser.go

@@ -116,6 +116,24 @@ func (p *ParsedProtobuf) Normalize() schema.ParsedSchema {
 	}
 }
 
+// HasTopLevelField reports whether any top-level message in the Protobuf
+// schema contains a field with the given name.
+func (p *ParsedProtobuf) HasTopLevelField(field string) bool {
+	if p.descriptor == nil {
+		return false
+	}
+	msgs := p.descriptor.Messages()
+	for i := 0; i < msgs.Len(); i++ {
+		fields := msgs.Get(i).Fields()
+		for j := 0; j < fields.Len(); j++ {
+			if string(fields.Get(j).Name()) == field {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // FormattedString returns the schema in the requested format.
 // Supported formats: "serialized" (base64-encoded FileDescriptorProto), "default" (canonical).
 func (p *ParsedProtobuf) FormattedString(format string) string {

internal/schema/types.go

@@ -27,6 +27,12 @@ type ParsedSchema interface {
 	// Normalize returns a normalized copy of this schema with deterministic
 	// representation for deduplication and comparison purposes.
 	Normalize() ParsedSchema
+
+	// HasTopLevelField reports whether the schema contains a top-level field
+	// with the given name. For Avro records this checks record fields, for
+	// Protobuf it checks fields across all top-level messages, and for JSON
+	// Schema it checks the "properties" object.
+	HasTopLevelField(field string) bool
 }
 
 // Parser is the interface for schema parsers.

internal/storage/cassandra/migrations.go

@@ -229,6 +229,10 @@ func Migrate(session *gocql.Session, keyspace string) error {
 		// subject_configs and global_config: compatibility_group
 		fmt.Sprintf(`ALTER TABLE %s.subject_configs ADD compatibility_group text`, qident(keyspace)),
 		fmt.Sprintf(`ALTER TABLE %s.global_config ADD compatibility_group text`, qident(keyspace)),
+
+		// subject_configs and global_config: validate_fields
+		fmt.Sprintf(`ALTER TABLE %s.subject_configs ADD validate_fields boolean`, qident(keyspace)),
+		fmt.Sprintf(`ALTER TABLE %s.global_config ADD validate_fields boolean`, qident(keyspace)),
 	}
 	for _, stmt := range alterStmts {
 		if err := session.Query(stmt).Exec(); err != nil {

internal/storage/cassandra/store.go

@@ -1361,13 +1361,13 @@ func (s *Store) GetConfig(ctx context.Context, subject string) (*storage.ConfigR
 		return s.GetGlobalConfig(ctx)
 	}
 	var compat, alias string
-	var normalize *bool
+	var normalize, validateFields *bool
 	var compatibilityGroup string
 	var defaultMetadataStr, overrideMetadataStr, defaultRulesetStr, overrideRulesetStr string
 	err := s.readQuery(
-		fmt.Sprintf(`SELECT compatibility, alias, normalize, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group FROM %s.subject_configs WHERE subject = ?`, qident(s.cfg.Keyspace)),
+		fmt.Sprintf(`SELECT compatibility, alias, normalize, validate_fields, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group FROM %s.subject_configs WHERE subject = ?`, qident(s.cfg.Keyspace)),
 		subject,
-	).WithContext(ctx).Scan(&compat, &alias, &normalize, &defaultMetadataStr, &overrideMetadataStr, &defaultRulesetStr, &overrideRulesetStr, &compatibilityGroup)
+	).WithContext(ctx).Scan(&compat, &alias, &normalize, &validateFields, &defaultMetadataStr, &overrideMetadataStr, &defaultRulesetStr, &overrideRulesetStr, &compatibilityGroup)
 	if err != nil {
 		if errors.Is(err, gocql.ErrNotFound) {
 			return nil, storage.ErrNotFound
@@ -1379,6 +1379,7 @@ func (s *Store) GetConfig(ctx context.Context, subject string) (*storage.ConfigR
 		CompatibilityLevel: compat,
 		Alias:              alias,
 		Normalize:          normalize,
+		ValidateFields:     validateFields,
 		CompatibilityGroup: compatibilityGroup,
 		DefaultMetadata:    unmarshalJSONText[storage.Metadata](defaultMetadataStr),
 		OverrideMetadata:   unmarshalJSONText[storage.Metadata](overrideMetadataStr),
@@ -1394,9 +1395,9 @@ func (s *Store) SetConfig(ctx context.Context, subject string, config *storage.C
 	}
 	compat := normalizeCompat(config.CompatibilityLevel)
 	return s.writeQuery(
-		fmt.Sprintf(`INSERT INTO %s.subject_configs (subject, compatibility, alias, normalize, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
-		subject, compat, config.Alias, config.Normalize,
+		fmt.Sprintf(`INSERT INTO %s.subject_configs (subject, compatibility, alias, normalize, validate_fields, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
+		subject, compat, config.Alias, config.Normalize, config.ValidateFields,
 		marshalJSONText(config.DefaultMetadata), marshalJSONText(config.OverrideMetadata),
 		marshalJSONText(config.DefaultRuleSet), marshalJSONText(config.OverrideRuleSet),
 		config.CompatibilityGroup,
@@ -1422,13 +1423,13 @@ func (s *Store) DeleteConfig(ctx context.Context, subject string) error {
 // GetGlobalConfig retrieves the global compatibility config.
 func (s *Store) GetGlobalConfig(ctx context.Context) (*storage.ConfigRecord, error) {
 	var compat, alias string
-	var normalize *bool
+	var normalize, validateFields *bool
 	var compatibilityGroup string
 	var defaultMetadataStr, overrideMetadataStr, defaultRulesetStr, overrideRulesetStr string
 	err := s.readQuery(
-		fmt.Sprintf(`SELECT compatibility, alias, normalize, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group FROM %s.global_config WHERE key = ?`, qident(s.cfg.Keyspace)),
+		fmt.Sprintf(`SELECT compatibility, alias, normalize, validate_fields, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group FROM %s.global_config WHERE key = ?`, qident(s.cfg.Keyspace)),
 		"global",
-	).WithContext(ctx).Scan(&compat, &alias, &normalize, &defaultMetadataStr, &overrideMetadataStr, &defaultRulesetStr, &overrideRulesetStr, &compatibilityGroup)
+	).WithContext(ctx).Scan(&compat, &alias, &normalize, &validateFields, &defaultMetadataStr, &overrideMetadataStr, &defaultRulesetStr, &overrideRulesetStr, &compatibilityGroup)
 	if err != nil {
 		if errors.Is(err, gocql.ErrNotFound) {
 			return nil, storage.ErrNotFound
@@ -1440,6 +1441,7 @@ func (s *Store) GetGlobalConfig(ctx context.Context) (*storage.ConfigRecord, err
 		CompatibilityLevel: compat,
 		Alias:              alias,
 		Normalize:          normalize,
+		ValidateFields:     validateFields,
 		CompatibilityGroup: compatibilityGroup,
 		DefaultMetadata:    unmarshalJSONText[storage.Metadata](defaultMetadataStr),
 		OverrideMetadata:   unmarshalJSONText[storage.Metadata](overrideMetadataStr),
@@ -1452,24 +1454,25 @@ func (s *Store) GetGlobalConfig(ctx context.Context) (*storage.ConfigRecord, err
 func (s *Store) SetGlobalConfig(ctx context.Context, config *storage.ConfigRecord) error {
 	compat := "BACKWARD"
 	var alias string
-	var normalize *bool
+	var normalize, validateFields *bool
 	var compatibilityGroup string
 	var defaultMetadata, overrideMetadata *storage.Metadata
 	var defaultRuleSet, overrideRuleSet *storage.RuleSet
 	if config != nil {
 		compat = normalizeCompat(config.CompatibilityLevel)
 		alias = config.Alias
 		normalize = config.Normalize
+		validateFields = config.ValidateFields
 		compatibilityGroup = config.CompatibilityGroup
 		defaultMetadata = config.DefaultMetadata
 		overrideMetadata = config.OverrideMetadata
 		defaultRuleSet = config.DefaultRuleSet
 		overrideRuleSet = config.OverrideRuleSet
 	}
 	return s.writeQuery(
-		fmt.Sprintf(`INSERT INTO %s.global_config (key, compatibility, alias, normalize, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
-		"global", compat, alias, normalize,
+		fmt.Sprintf(`INSERT INTO %s.global_config (key, compatibility, alias, normalize, validate_fields, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
+		"global", compat, alias, normalize, validateFields,
 		marshalJSONText(defaultMetadata), marshalJSONText(overrideMetadata),
 		marshalJSONText(defaultRuleSet), marshalJSONText(overrideRuleSet),
 		compatibilityGroup,
@@ -1479,9 +1482,9 @@ func (s *Store) SetGlobalConfig(ctx context.Context, config *storage.ConfigRecor
 // DeleteGlobalConfig resets the global config to the default (BACKWARD).
 func (s *Store) DeleteGlobalConfig(ctx context.Context) error {
 	return s.writeQuery(
-		fmt.Sprintf(`INSERT INTO %s.global_config (key, compatibility, alias, normalize, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
-		"global", "BACKWARD", "", nil, "", "", "", "", "",
+		fmt.Sprintf(`INSERT INTO %s.global_config (key, compatibility, alias, normalize, validate_fields, default_metadata, override_metadata, default_ruleset, override_ruleset, compatibility_group, updated_at)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now())`, qident(s.cfg.Keyspace)),
+		"global", "BACKWARD", "", nil, nil, "", "", "", "", "",
 	).WithContext(ctx).Exec()
 }
 

internal/storage/mysql/migrations.go

@@ -117,4 +117,7 @@ var migrations = []string{
 
 	// Migration 19: Add compatibility_group column to configs
 	"ALTER TABLE configs ADD COLUMN compatibility_group VARCHAR(255)",
+
+	// Migration 20: Add validate_fields column to configs
+	"ALTER TABLE configs ADD COLUMN validate_fields BOOLEAN",
 }