confluent: add default_schema_id field to the schema_registry_decode processor

Author: mmatczuk

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 
 ## 4.64.0 - TBD
 
+### Added
+
+- Added `default_schema_id` field to the `schema_registry_decode` processor. (@mmatczuk)
+
 ### Changed
 
 - (google_cloud_storage) Field `bucket` can now be interpolated (@rockwotj)

docs/modules/components/pages/processors/schema_registry_decode.adoc

@@ -48,6 +48,7 @@ schema_registry_decode:
     emit_default_values: false
   cache_duration: 10m
   url: "" # No default (required)
+  default_schema_id: 0 # No default (optional)
 ```
 
 --
@@ -86,6 +87,7 @@ schema_registry_decode:
     emit_default_values: false
   cache_duration: 10m
   url: "" # No default (required)
+  default_schema_id: 0 # No default (optional)
   oauth:
     enabled: false
     consumer_key: ""
@@ -338,6 +340,14 @@ The base URL of the schema registry service.
 *Type*: `string`
 
 
+=== `default_schema_id`
+
+If set, this schema ID will be used when a message's schema header cannot be read (ErrBadHeader). If not set, schema header errors will be returned.
+
+
+*Type*: `int`
+
+
 === `oauth`
 
 Allows you to specify open authentication via OAuth version 1.

internal/impl/confluent/processor_schema_registry_decode.go

@@ -179,7 +179,10 @@ root = this.apply("debeziumTimestampToAvroTimestamp")
 				Description("The duration after which a schema is considered stale and will be removed from the cache.").
 				Default("10m").Example("1h").Example("5m"),
 		).
-		Field(service.NewURLField("url").Description("The base URL of the schema registry service."))
+		Field(service.NewURLField("url").Description("The base URL of the schema registry service.")).
+		Field(service.NewIntField("default_schema_id").
+			Description("If set, this schema ID will be used when a message's schema header cannot be read (ErrBadHeader). If not set, schema header errors will be returned.").
+			Optional())
 
 	for _, f := range service.NewHTTPRequestAuthSignerFields() {
 		spec = spec.Field(f.Version("4.7.0"))
@@ -212,6 +215,7 @@ type decodingConfig struct {
 		emitUnpopulated   bool
 		emitDefaultValues bool
 	}
+	defaultSchemaID int
 }
 
 type schemaRegistryDecoder struct {
@@ -287,6 +291,14 @@ func newSchemaRegistryDecoderFromConfig(conf *service.ParsedConfig, mgr *service
 	if err != nil {
 		return nil, err
 	}
+
+	if conf.Contains("default_schema_id") {
+		cfg.defaultSchemaID, err = conf.FieldInt("default_schema_id")
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	cacheDuration, err := conf.FieldDuration("cache_duration")
 	if err != nil {
 		return nil, err
@@ -335,7 +347,11 @@ func (s *schemaRegistryDecoder) Process(_ context.Context, msg *service.Message)
 
 	var ch franz_sr.ConfluentHeader
 	id, remaining, err := ch.DecodeID(b)
-	if err != nil {
+	if errors.Is(err, franz_sr.ErrBadHeader) && s.cfg.defaultSchemaID != 0 {
+		// Use default schema ID when header cannot be read
+		id = s.cfg.defaultSchemaID
+		remaining = b
+	} else if err != nil {
 		return nil, err
 	}
 

internal/impl/confluent/processor_schema_registry_decode_test.go

@@ -680,6 +680,81 @@ func TestSchemaRegistryDecodeProtobuf(t *testing.T) {
 	decoder.cacheMut.Unlock()
 }
 
+func TestSchemaRegistryDecodeWithDefaultSchemaID(t *testing.T) {
+	urlStr := runSchemaRegistryServer(t, func(path string) ([]byte, error) {
+		if path == "/schemas/ids/3" {
+			return mustJBytes(t, map[string]any{
+				"schema": testSchema,
+			}), nil
+		}
+		return nil, nil
+	})
+
+	tests := []struct {
+		name        string
+		input       string
+		output      string
+		defaultID   int
+		errContains string
+	}{
+		{
+			name:        "error when no default schema is set",
+			input:       "\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing", // Invalid header
+			errContains: "5 byte header for value is missing or does not have 0 magic byte",
+		},
+		{
+			name:        "different error doesn't use default schema",
+			input:       "\x00\x00\x00\x00\x09", // Valid header but non-existent schema
+			defaultID:   3,
+			errContains: "schema 9 not found by registry: not found",
+		},
+		{
+			name:      "no header uses default schema",
+			input:     "\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing", // No valid header at all
+			output:    `{"Address":{"my.namespace.com.address":{"City":{"string":"foo"},"State":"bar"}},"MaybeHobby":{"string":"dancing"},"Name":"foo"}`,
+			defaultID: 3,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			cfg := decodingConfig{}
+			cfg.avro.rawUnions = false
+			if test.defaultID != 0 {
+				cfg.defaultSchemaID = test.defaultID
+			}
+
+			decoder, err := newSchemaRegistryDecoder(urlStr, noopReqSign, nil, cfg, schemaStaleAfter, service.MockResources())
+			require.NoError(t, err)
+			defer func() {
+				require.NoError(t, decoder.Close(t.Context()))
+			}()
+
+			outMsgs, err := decoder.Process(t.Context(), service.NewMessage([]byte(test.input)))
+			if test.errContains != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), test.errContains)
+			} else {
+				require.NoError(t, err)
+				require.Len(t, outMsgs, 1)
+
+				b, err := outMsgs[0].AsBytes()
+				require.NoError(t, err)
+
+				jdopts := jsondiff.DefaultJSONOptions()
+				diff, explanation := jsondiff.Compare(b, []byte(test.output), &jdopts)
+				assert.JSONEq(t, test.output, string(b))
+				assert.Equalf(t, jsondiff.FullMatch.String(), diff.String(), "%s: %s", test.name, explanation)
+
+				v, ok := outMsgs[0].MetaGetMut("schema_id")
+				assert.True(t, ok)
+				assert.Equal(t, test.defaultID, v)
+			}
+		})
+	}
+}
+
 func TestSchemaRegistryDecodeJson(t *testing.T) {
 	returnedSchema3 := false
 	urlStr := runSchemaRegistryServer(t, func(path string) ([]byte, error) {