[UX-1513] Schema Registry: context aware schema resolution

backend/pkg/api/connect/service/console/mapper.go

@@ -62,8 +62,9 @@ func rpcPublishMessagePayloadOptionsToSerializeInput(po *v1alpha.PublishMessageP
 	}
 
 	input := &serde.RecordPayloadInput{
-		Payload:  po.GetData(),
-		Encoding: encoding,
+		Payload:       po.GetData(),
+		Encoding:      encoding,
+		SchemaContext: po.GetSchemaContext(),
 	}
 
 	if po.GetSchemaId() > 0 {

backend/pkg/api/connect/service/console/service.go

@@ -28,6 +28,7 @@ import (
 	"github.com/redpanda-data/console/backend/pkg/console"
 	v1alpha "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
 	dataplane "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/dataplane/v1alpha2"
+	schemacache "github.com/redpanda-data/console/backend/pkg/schema"
 )
 
 // Service that implements the ConsoleServiceHandler interface.
@@ -97,6 +98,7 @@ func (api *Service) ListMessages(
 		IgnoreMaxSizeLimit:    req.Msg.GetIgnoreMaxSizeLimit(),
 		KeyDeserializer:       fromProtoEncoding(req.Msg.GetKeyDeserializer()),
 		ValueDeserializer:     fromProtoEncoding(req.Msg.GetValueDeserializer()),
+		SchemaContext:         req.Msg.GetSchemaContext(),
 		PageToken:             lmq.PageToken,
 		PageSize:              int(req.Msg.GetPageSize()),
 	}
@@ -229,6 +231,11 @@ func (api *Service) GenerateSchemaSample(
 		indexPath = append(indexPath, int(v))
 	}
 
+	// Resolve the schema ID in the context the client named.
+	if schemaCtx := req.Msg.GetSchemaContext(); schemaCtx != "" {
+		ctx = schemacache.InContext(ctx, schemaCtx)
+	}
+
 	sample, err := api.consoleSvc.GenerateSchemaSampleJSON(ctx, int(req.Msg.GetSchemaId()), indexPath)
 	if err != nil {
 		return nil, apierrors.NewConnectError(

backend/pkg/console/list_messages.go

@@ -60,6 +60,7 @@ type ListMessageRequest struct {
 	IgnoreMaxSizeLimit    bool
 	KeyDeserializer       serde.PayloadEncoding
 	ValueDeserializer     serde.PayloadEncoding
+	SchemaContext         string // Empty resolves the topic's context, "." forces the default.
 
 	// Pagination fields (used when PageSize > 0)
 	PageToken string
@@ -133,6 +134,7 @@ type TopicConsumeRequest struct {
 	KeyDeserializer       serde.PayloadEncoding
 	ValueDeserializer     serde.PayloadEncoding
 	Direction             string // "desc" or "asc" - used for message ordering
+	SchemaContext         string // Empty means default.
 }
 
 // ListMessages processes a list message request as sent from the Frontend. This function is responsible (mostly
@@ -167,6 +169,12 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 		return fmt.Errorf("failed to get metadata for topic %s: %w", listReq.TopicName, topicMetadata.Err)
 	}
 
+	// Schema IDs are context-local; an explicit context wins over the topic's.
+	schemaCtx := listReq.SchemaContext
+	if schemaCtx == "" {
+		schemaCtx = s.resolveTopicSchemaContext(ctx, adminCl, listReq.TopicName)
+	}
+
 	partitionByID := make(map[int32]kadm.PartitionDetail)
 	onlinePartitionIDs := make([]int32, 0)
 	offlinePartitionIDs := make([]int32, 0)
@@ -263,6 +271,7 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 		KeyDeserializer:       listReq.KeyDeserializer,
 		ValueDeserializer:     listReq.ValueDeserializer,
 		Direction:             direction,
+		SchemaContext:         schemaCtx,
 	}
 
 	progress.OnPhase("Consuming messages")
@@ -1015,6 +1024,7 @@ func (s *Service) startMessageWorker(ctx context.Context, wg *sync.WaitGroup,
 				IgnoreMaxSizeLimit: consumeReq.IgnoreMaxSizeLimit,
 				KeyEncoding:        consumeReq.KeyDeserializer,
 				ValueEncoding:      consumeReq.ValueDeserializer,
+				SchemaContext:      consumeReq.SchemaContext,
 			})
 
 		headersByKey := make(map[string][]byte, len(deserializedRec.Headers))

backend/pkg/console/produce_records.go

@@ -32,10 +32,16 @@ func (s *Service) ProduceRecord(
 	useTransactions bool,
 	compressionOpts []kgo.CompressionCodec,
 ) (*ProduceRecordResponse, error) {
+	_, adminCl, err := s.kafkaClientFactory.GetKafkaClient(ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	data, err := s.serdeSvc.SerializeRecord(ctx, serde.SerializeInput{
-		Topic: topic,
-		Key:   *key,
-		Value: *value,
+		Topic:         topic,
+		SchemaContext: s.resolveTopicSchemaContext(ctx, adminCl, topic),
+		Key:           *key,
+		Value:         *value,
 	})
 	if err != nil {
 		return &ProduceRecordResponse{

backend/pkg/console/schema_context.go

@@ -0,0 +1,67 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package console
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/twmb/franz-go/pkg/kadm"
+
+	schemacache "github.com/redpanda-data/console/backend/pkg/schema"
+)
+
+// topicSchemaContext returns the topic's redpanda.schema.registry.context, or "".
+func topicSchemaContext(configs []kadm.Config) string {
+	for _, c := range configs {
+		if c.Key == schemacache.TopicConfigSchemaRegistryContext {
+			return schemacache.NormalizeContextName(c.MaybeValue())
+		}
+	}
+	return ""
+}
+
+// schemaContextForTopic reads the topic's schema registry context via DescribeConfigs.
+func schemaContextForTopic(ctx context.Context, adminCl *kadm.Client, topic string) (string, error) {
+	rcs, err := adminCl.DescribeTopicConfigs(ctx, topic)
+	if err != nil {
+		return "", err
+	}
+	rc, err := rcs.On(topic, nil)
+	if err != nil {
+		return "", err
+	}
+	if rc.Err != nil {
+		return "", rc.Err
+	}
+	return topicSchemaContext(rc.Configs), nil
+}
+
+// resolveTopicSchemaContext returns the topic's schema registry context. Lookup
+// failures are logged and fall back to the default context.
+func (s *Service) resolveTopicSchemaContext(ctx context.Context, adminCl *kadm.Client, topic string) string {
+	if !s.cfg.SchemaRegistry.Enabled {
+		return ""
+	}
+
+	schemaCtx, err := schemaContextForTopic(ctx, adminCl, topic)
+	if err != nil {
+		s.logger.WarnContext(ctx, "failed to determine the topic's schema registry context, using the default context",
+			slog.String("topic", topic),
+			slog.Any("error", err))
+		return ""
+	}
+	if schemaCtx != "" {
+		s.logger.DebugContext(ctx, "resolving schemas in the topic's schema registry context",
+			slog.String("topic", topic),
+			slog.String("schema_context", schemaCtx))
+	}
+	return schemaCtx
+}

backend/pkg/console/schema_context_test.go

@@ -0,0 +1,67 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package console
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/twmb/franz-go/pkg/kadm"
+)
+
+func TestTopicSchemaContext(t *testing.T) {
+	tests := []struct {
+		name     string
+		configs  []kadm.Config
+		expected string
+	}{
+		{name: "nil configs", configs: nil, expected: ""},
+		{
+			name:     "context not set",
+			configs:  []kadm.Config{{Key: "cleanup.policy", Value: new("delete")}},
+			expected: "",
+		},
+		{
+			name:     "context set",
+			configs:  []kadm.Config{{Key: "redpanda.schema.registry.context", Value: new(".prod")}},
+			expected: ".prod",
+		},
+		{
+			name:     "context present without value",
+			configs:  []kadm.Config{{Key: "redpanda.schema.registry.context", Value: nil}},
+			expected: "",
+		},
+		{
+			name:     "explicit default context",
+			configs:  []kadm.Config{{Key: "redpanda.schema.registry.context", Value: new(".")}},
+			expected: "",
+		},
+		{
+			name:     "context without leading dot is normalized",
+			configs:  []kadm.Config{{Key: "redpanda.schema.registry.context", Value: new("prod")}},
+			expected: ".prod",
+		},
+		{
+			name: "context among other configs",
+			configs: []kadm.Config{
+				{Key: "cleanup.policy", Value: new("delete")},
+				{Key: "redpanda.schema.registry.context", Value: new(".staging")},
+				{Key: "redpanda.iceberg.mode", Value: new("value_schema_id_prefix")},
+			},
+			expected: ".staging",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, topicSchemaContext(tc.configs))
+		})
+	}
+}

backend/pkg/console/schema_registry.go

@@ -27,6 +27,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/redpanda-data/console/backend/pkg/proto"
+	schemacache "github.com/redpanda-data/console/backend/pkg/schema"
 )
 
 // SchemaRegistryMode returns the schema registry mode.
@@ -441,7 +442,8 @@ func (s *Service) GetSchemaRegistrySubjectDetails(ctx context.Context, subjectNa
 		return nil, err
 	}
 
-	s.populateProtoMessageTypes(ctx, subjectName, schemas)
+	// Resolve the message types in the context of the qualified subject.
+	s.populateProtoMessageTypes(schemacache.InContext(ctx, schemacache.ContextFromSubject(subjectName)), subjectName, schemas)
 
 	var schemaType sr.SchemaType
 	if len(schemas) > 0 {
@@ -473,8 +475,9 @@ func (s *Service) populateProtoMessageTypes(ctx context.Context, subjectName str
 			if err != nil {
 				s.logger.WarnContext(grpCtx, "failed to resolve protobuf message types",
 					slog.String("subject", subjectName),
-					slog.Int("schemaId", schemas[i].ID),
-					slog.Any("err", err))
+					slog.Int("schema_id", schemas[i].ID),
+					slog.String("schema_context", schemacache.ContextName(grpCtx)),
+					slog.Any("error", err))
 				return nil
 			}
 			schemas[i].MessageTypes = types
@@ -960,24 +963,80 @@ func (s *Service) GetSchemaUsagesByID(ctx context.Context, schemaID int, subject
 		return nil, err
 	}
 
-	callCtx := ctx
 	if subject != "" {
-		callCtx = sr.WithParams(ctx, sr.Subject(subject))
+		// With a subject the registry searches across contexts itself.
+		res, err := srClient.SchemaUsagesByID(sr.WithParams(ctx, sr.Subject(subject)), schemaID)
+		if err != nil {
+			return nil, err
+		}
+		return mapSchemaVersions(res), nil
 	}
-	res, err := srClient.SchemaUsagesByID(callCtx, schemaID)
-	if err != nil {
-		return nil, err
+
+	return schemaUsagesAcrossContexts(ctx, srClient, schemaID)
+}
+
+// schemaUsagesAcrossContexts looks up the subjects using schemaID in every
+// context, since schema IDs are context-local.
+func schemaUsagesAcrossContexts(ctx context.Context, srClient *rpsr.Client, schemaID int) ([]SchemaVersion, error) {
+	// Always search the default context; registries without contexts fail /contexts.
+	contextNames := []string{""}
+	if contexts, err := srClient.Contexts(ctx); err == nil {
+		for _, name := range contexts {
+			if normalized := schemacache.NormalizeContextName(name); normalized != "" {
+				contextNames = append(contextNames, normalized)
+			}
+		}
+	}
+
+	results := make([][]sr.SubjectSchema, len(contextNames))
+	errs := make([]error, len(contextNames))
+	grp, grpCtx := errgroup.WithContext(ctx)
+	grp.SetLimit(10)
+	for i, name := range contextNames {
+		grp.Go(func() error {
+			callCtx := grpCtx
+			if name != "" {
+				callCtx = sr.InContext(grpCtx, name)
+			}
+			results[i], errs[i] = srClient.SchemaUsagesByID(callCtx, schemaID)
+			return nil
+		})
+	}
+	_ = grp.Wait()
+
+	// De-duplicate: a subject may be reported by more than one lookup.
+	seen := make(map[SchemaVersion]struct{})
+	schemaVersions := make([]SchemaVersion, 0)
+	for _, res := range results {
+		for _, sv := range mapSchemaVersions(res) {
+			if _, ok := seen[sv]; ok {
+				continue
+			}
+			seen[sv] = struct{}{}
+			schemaVersions = append(schemaVersions, sv)
+		}
+	}
+	if len(schemaVersions) == 0 {
+		// Unknown everywhere: return the default context's error.
+		for _, err := range errs {
+			if err != nil {
+				return nil, err
+			}
+		}
 	}
 
+	return schemaVersions, nil
+}
+
+func mapSchemaVersions(res []sr.SubjectSchema) []SchemaVersion {
 	schemaVersions := make([]SchemaVersion, len(res))
 	for i, r := range res {
 		schemaVersions[i] = SchemaVersion{
 			Subject: r.Subject,
 			Version: r.Version,
 		}
 	}
-
-	return schemaVersions, nil
+	return schemaVersions
 }
 
 // CheckSchemaRegistryACLSupport checks if the Schema Registry supports ACL