Fix schema_registry_decode processor

When dealing with logical types, we can't serialise the native
struct emitted by goavro directly to JSON, since that will discard
schema information that `codec.TextualFromNative` uses to produce
the expected JSON.

Also, the tip version of goavro is required because the fix for
this regression linkedin/goavro#242 was
merged, but there isn't a tagged release yet.

Fixes redpanda-data#1161.

go.mod

@@ -73,7 +73,7 @@ require (
 	github.com/jmespath/go-jmespath v0.4.0
 	github.com/klauspost/compress v1.15.1 // indirect
 	github.com/lib/pq v1.10.4
-	github.com/linkedin/goavro/v2 v2.11.0
+	github.com/linkedin/goavro/v2 v2.11.1-0.20220404183248-ee3a1f1d6e9c
 	github.com/matoous/go-nanoid/v2 v2.0.0
 	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/microcosm-cc/bluemonday v1.0.17

go.sum

@@ -661,8 +661,8 @@ github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.10.4 h1:SO9z7FRPzA03QhHKJrH5BXA6HU1rS4V2nIVrrNC1iYk=
 github.com/lib/pq v1.10.4/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/linkedin/goavro/v2 v2.9.8/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
-github.com/linkedin/goavro/v2 v2.11.0 h1:AlU/NR32ESbC/dlzbhTjyqybwESupUCc3SrrHg2qdTg=
-github.com/linkedin/goavro/v2 v2.11.0/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
+github.com/linkedin/goavro/v2 v2.11.1-0.20220404183248-ee3a1f1d6e9c h1:TyHvqhsZvEkiKAZb9P0jFt6XUGh7HLSD/mp8juKDgGg=
+github.com/linkedin/goavro/v2 v2.11.1-0.20220404183248-ee3a1f1d6e9c/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=

internal/impl/confluent/processor_schema_registry_decode.go

@@ -87,7 +87,7 @@ func newSchemaRegistryDecoderFromConfig(conf *service.ParsedConfig, logger *serv
 	if err != nil {
 		return nil, err
 	}
-	return newSchemaRegistryDecoder(urlStr, tlsConf, false, logger)
+	return newSchemaRegistryDecoder(urlStr, tlsConf, true, logger)
 }
 
 func newSchemaRegistryDecoder(urlStr string, tlsConf *tls.Config, avroRawJSON bool, logger *service.Logger) (*schemaRegistryDecoder, error) {

internal/impl/confluent/processor_schema_registry_decode_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nsf/jsondiff"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -103,6 +104,59 @@ const testSchema = `{
 	]
 }`
 
+const testSchemaLogicalTypes = `{
+	"type": "record",
+	"name": "LogicalTypes",
+	"fields": [
+		{
+			"default": null,
+			"name": "int_time_millis",
+			"type": [
+				"null",
+				{
+					"type": "int",
+					"logicalType": "time-millis"
+				}
+			]
+		},
+		{
+			"default": null,
+			"name": "long_time_micros",
+			"type": [
+				"null",
+				{
+					"type": "long",
+					"logicalType": "time-micros"
+				}
+			]
+		},
+		{
+			"default": null,
+			"name": "long_timestamp_micros",
+			"type": [
+				"null",
+				{
+					"type": "long",
+					"logicalType": "timestamp-micros"
+				}
+			]
+		},
+		{
+			"default": null,
+			"name": "pos_0_33333333",
+			"type": [
+				"null",
+				{
+					"logicalType": "decimal",
+					"precision": 16,
+					"scale": 2,
+					"type": "bytes"
+				}
+			]
+		}
+	]
+}`
+
 func TestSchemaRegistryDecodeAvro(t *testing.T) {
 	payload3, err := json.Marshal(struct {
 		Schema string `json:"schema"`
@@ -111,20 +165,29 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 	})
 	require.NoError(t, err)
 
+	payload4, err := json.Marshal(struct {
+		Schema string `json:"schema"`
+	}{
+		Schema: testSchemaLogicalTypes,
+	})
+	require.NoError(t, err)
+
 	returnedSchema3 := false
 	urlStr := runSchemaRegistryServer(t, func(path string) ([]byte, error) {
 		switch path {
 		case "/schemas/ids/3":
 			assert.False(t, returnedSchema3)
 			returnedSchema3 = true
 			return payload3, nil
+		case "/schemas/ids/4":
+			return payload4, nil
 		case "/schemas/ids/5":
 			return nil, fmt.Errorf("nope")
 		}
 		return nil, nil
 	})
 
-	decoder, err := newSchemaRegistryDecoder(urlStr, nil, false, nil)
+	decoder, err := newSchemaRegistryDecoder(urlStr, nil, true, nil)
 	require.NoError(t, err)
 
 	tests := []struct {
@@ -143,15 +206,20 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 			input:  "\x00\x00\x00\x00\x03\x06foo\x02\x06foo\x06bar\x00",
 			output: `{"Address":{"my.namespace.com.address":{"City":"foo","State":"bar"}},"MaybeHobby":null,"Name":"foo"}`,
 		},
+		{
+			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":"!"}}`,
+		},
 		{
 			name:        "non-empty magic byte",
 			input:       "\x06\x00\x00\x00\x03\x06foo\x02\x06foo\x06bar",
 			errContains: "version number 6 not supported",
 		},
 		{
 			name:        "non-existing schema",
-			input:       "\x00\x00\x00\x00\x04\x06foo\x02\x06foo\x06bar",
-			errContains: "schema '4' not found by registry",
+			input:       "\x00\x00\x00\x00\x06\x06foo\x02\x06foo\x06bar",
+			errContains: "schema '6' not found by registry",
 		},
 		{
 			name:        "server fails",
@@ -173,9 +241,11 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 
 				b, err := outMsgs[0].AsBytes()
 				require.NoError(t, err)
-				assert.Equal(t, test.output, string(b))
-			}
 
+				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)
+			}
 		})
 	}
 
@@ -190,7 +260,7 @@ func TestSchemaRegistryDecodeClearExpired(t *testing.T) {
 		return nil, fmt.Errorf("nope")
 	})
 
-	decoder, err := newSchemaRegistryDecoder(urlStr, nil, false, nil)
+	decoder, err := newSchemaRegistryDecoder(urlStr, nil, true, nil)
 	require.NoError(t, err)
 	require.NoError(t, decoder.Close(context.Background()))