Merge pull request #3142 from rockwotj/schema_id

Author: rockwotj

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 - Field `instance_id` added to `kafka`, `kafka_franz`, `ockam_kafka`, `redpanda`, `redpanda_common`, and `redpanda_migrator` inputs. (@rockwotj)
 - Fields `rebalance_timeout`, `session_timeout` and `heartbeat_interval` added to the `kafka_franz`, `redpanda`, `redpanda_common`, `redpanda_migrator` and `ockam_kafka` inputs. (@rockwotj)
 - Field `avro.preserve_logical_types` for processor `schema_registry_decode` was added to preserve logical types instead of decoding them as their primitive representation. (@rockwotj)
+- Processor `schema_registry_decode` now adds metadata `schema_id` for the schema's ID in the schema registry. (@rockwotj)
 
 ### Changed
 

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

@@ -105,6 +105,12 @@ However, it is possible to instead create documents in https://pkg.go.dev/github
 
 This processor decodes protobuf messages to JSON documents, you can read more about JSON mapping of protobuf messages here: https://developers.google.com/protocol-buffers/docs/proto3#json
 
+== Metadata
+
+This processor also adds the following metadata to each outgoing message:
+
+schema_id: the ID of the schema in the schema registry that was associated with the message.
+
 
 == Fields
 

internal/impl/confluent/processor_schema_registry_decode.go

@@ -60,6 +60,12 @@ However, it is possible to instead create documents in https://pkg.go.dev/github
 == Protobuf format
 
 This processor decodes protobuf messages to JSON documents, you can read more about JSON mapping of protobuf messages here: https://developers.google.com/protocol-buffers/docs/proto3#json
+
+== Metadata
+
+This processor also adds the following metadata to each outgoing message:
+
+schema_id: the ID of the schema in the schema registry that was associated with the message.
 `).
 		Field(service.NewBoolField("avro_raw_json").
 			Description("Whether Avro messages should be decoded into normal JSON (\"json that meets the expectations of regular internet json\") rather than https://avro.apache.org/docs/current/specification/_print/#json-encoding[Avro JSON^]. If `true` the schema returned from the subject should be decoded as https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull[standard json^] instead of as https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodec[avro json^]. There is a https://github.com/linkedin/goavro/blob/5ec5a5ee7ec82e16e6e2b438d610e1cab2588393/union.go#L224-L249[comment in goavro^], the https://github.com/linkedin/goavro[underlining library used for avro serialization^], that explains in more detail the difference between the standard json and avro json.").
@@ -209,6 +215,7 @@ func (s *schemaRegistryDecoder) Process(ctx context.Context, msg *service.Messag
 	if err := decoder(msg); err != nil {
 		return nil, err
 	}
+	msg.MetaSetMut("schema_id", id)
 
 	return service.MessageBatch{msg}, nil
 }

internal/impl/confluent/processor_schema_registry_decode_test.go

@@ -263,28 +263,33 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 	})
 
 	tests := []struct {
+		schemaID    int
 		name        string
 		input       string
 		output      string
 		hambaOutput string
 		errContains string
 	}{
 		{
-			name:   "successful message",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing",
-			output: `{"Address":{"my.namespace.com.address":{"City":{"string":"foo"},"State":"bar"}},"MaybeHobby":{"string":"dancing"},"Name":"foo"}`,
+			schemaID: 3,
+			name:     "successful message",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing",
+			output:   `{"Address":{"my.namespace.com.address":{"City":{"string":"foo"},"State":"bar"}},"MaybeHobby":{"string":"dancing"},"Name":"foo"}`,
 		},
 		{
-			name:   "successful message with null hobby",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x00",
-			output: `{"Address":{"my.namespace.com.address":{"City":{"string":"foo"},"State":"bar"}},"MaybeHobby":null,"Name":"foo"}`,
+			schemaID: 3,
+			name:     "successful message with null hobby",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x00",
+			output:   `{"Address":{"my.namespace.com.address":{"City":{"string":"foo"},"State":"bar"}},"MaybeHobby":null,"Name":"foo"}`,
 		},
 		{
-			name:   "successful message no address and null hobby",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x00\x00",
-			output: `{"Name":"foo","MaybeHobby":null,"Address": null}`,
+			schemaID: 3,
+			name:     "successful message no address and null hobby",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x00\x00",
+			output:   `{"Name":"foo","MaybeHobby":null,"Address": null}`,
 		},
 		{
+			schemaID:    4,
 			name:        "successful message with logical types",
 			input:       "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
 			output:      `{"int_time_millis":{"int.time-millis":35245000},"long_time_micros":{"long.time-micros":20192000000000},"long_timestamp_micros":{"long.timestamp-micros":62135596800000000},"pos_0_33333333":{"bytes.decimal":"!"}}`,
@@ -341,6 +346,10 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 				diff, explanation := jsondiff.Compare(b, []byte(output), &jdopts)
 				assert.JSONEq(t, 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.schemaID, v)
 			}
 		}
 		t.Run("hamba/"+test.name, func(t *testing.T) { fn(t, true) })
@@ -387,28 +396,33 @@ func TestSchemaRegistryDecodeAvroRawJson(t *testing.T) {
 	})
 
 	tests := []struct {
+		schemaID    int
 		name        string
 		input       string
 		output      string
 		hambaOutput string
 		errContains string
 	}{
 		{
-			name:   "successful message",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing",
-			output: `{"Address":{"City":"foo","State":"bar"},"Name":"foo","MaybeHobby":"dancing"}`,
+			schemaID: 3,
+			name:     "successful message",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x02\x0edancing",
+			output:   `{"Address":{"City":"foo","State":"bar"},"Name":"foo","MaybeHobby":"dancing"}`,
 		},
 		{
-			name:   "successful message with null hobby",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x00",
-			output: `{"Address":{"City":"foo","State":"bar"},"MaybeHobby":null,"Name":"foo"}`,
+			schemaID: 3,
+			name:     "successful message with null hobby",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x02\x02\x06foo\x06bar\x00",
+			output:   `{"Address":{"City":"foo","State":"bar"},"MaybeHobby":null,"Name":"foo"}`,
 		},
 		{
-			name:   "successful message no address and null hobby",
-			input:  "\x00\x00\x00\x00\x03\x06foo\x00\x00",
-			output: `{"Name":"foo","MaybeHobby":null,"Address": null}`,
+			schemaID: 3,
+			name:     "successful message no address and null hobby",
+			input:    "\x00\x00\x00\x00\x03\x06foo\x00\x00",
+			output:   `{"Name":"foo","MaybeHobby":null,"Address": null}`,
 		},
 		{
+			schemaID:    4,
 			name:        "successful message with logical types",
 			input:       "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
 			output:      `{"int_time_millis":35245000,"long_time_micros":20192000000000,"long_timestamp_micros":62135596800000000,"pos_0_33333333":"!"}`,
@@ -464,8 +478,11 @@ func TestSchemaRegistryDecodeAvroRawJson(t *testing.T) {
 				jdopts := jsondiff.DefaultJSONOptions()
 				diff, explanation := jsondiff.Compare(b, []byte(output), &jdopts)
 				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.schemaID, v)
+			}
 		}
 		t.Run("hamba/"+test.name, func(t *testing.T) { fn(t, true) })
 		t.Run("goavro/"+test.name, func(t *testing.T) { fn(t, false) })
@@ -567,8 +584,11 @@ func TestSchemaRegistryDecodeProtobuf(t *testing.T) {
 
 				b, err := outMsgs[0].AsBytes()
 				require.NoError(t, err)
-
 				assert.JSONEq(t, test.output, string(b), "%s: %s", test.name)
+
+				v, ok := outMsgs[0].MetaGetMut("schema_id")
+				assert.True(t, ok)
+				assert.Equal(t, 1, v)
 			}
 		})
 	}
@@ -639,6 +659,9 @@ func TestSchemaRegistryDecodeJson(t *testing.T) {
 				jdopts := jsondiff.DefaultJSONOptions()
 				diff, explanation := jsondiff.Compare(b, []byte(test.output), &jdopts)
 				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, 3, v)
 			}
 		})
 	}