confluent,avro,salesforce: replace linkedin/goavro and hamba/avro with twmb/avro

Remove all linkedin/goavro and hamba/avro imports from the confluent,
avro, and salesforce packages. All Avro encoding and decoding now
uses github.com/twmb/avro.

Schema registry decoder uses EncodeJSON for JSON output paths and
Decode(TaggedUnions, TagLogicalTypes) with CustomType registrations
for the preserve_logical_types path. The avro_walker.go file contains
only connect-specific CustomType registrations: Kafka Connect type
translation (Debezium timestamps), Duration→Time for time-of-day
fields, and Duration→string for the duration logical type. All schema
tree walking, union wrapping, and branch naming are handled by
twmb/avro at decode time.

Avro processor and scanner use EncodeJSON(TaggedUnions, TagLogicalTypes)
for Avro JSON output and EncodeJSON() for standard JSON output.

Delete normalize_for_avro_schema.go (572 lines) — type coercion,
timestamp parsing, and union normalization are handled by twmb/avro's
Encode and DecodeJSON.

Subsumes PR #4124 (Fix avro timestamp encoding): twmb/avro's Encode
natively accepts RFC 3339 strings for timestamp fields.

File renames:
  serde_goavro.go → serde_avro.go
  serde_hamba_avro.go → avro_walker.go
  useHamba → preserveLogicalTypes

Breaking change in preserve_logical_types mode:
  Avro fixed fields in structured output (SetStructuredMut) are
  now base64-encoded strings instead of arrays of integers. This
  affects Bloblang expressions that access fixed field values as
  arrays (e.g. this.fixedField.index(0)). The change only affects
  users with preserve_logical_types enabled AND schemas containing
  fixed-type fields — a rare combination. Fixed fields are binary
  data and base64 is the standard JSON representation; the old
  integer-array format was a hamba-specific behavior.

Error messages from twmb/avro differ in wording from goavro/hamba
but all invalid inputs are still rejected.

Author: twmb

go.mod

@@ -113,7 +113,6 @@ require (
 	github.com/jackc/pgx/v5 v5.9.1
 	github.com/jhump/protoreflect v1.18.0
 	github.com/lib/pq v1.12.0
-	github.com/linkedin/goavro/v2 v2.15.0
 	github.com/matoous/go-nanoid/v2 v2.1.0
 	github.com/microcosm-cc/bluemonday v1.0.27
 	github.com/microsoft/go-mssqldb v1.9.8
@@ -167,6 +166,7 @@ require (
 	github.com/timeplus-io/proton-go-driver/v2 v2.1.4
 	github.com/tmc/langchaingo v0.1.14
 	github.com/trinodb/trino-go-client v0.333.0
+	github.com/twmb/avro v1.3.1
 	github.com/twmb/franz-go v1.20.7
 	github.com/twmb/franz-go/pkg/kadm v1.17.2
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0
@@ -276,6 +276,7 @@ require (
 	github.com/knadh/koanf/providers/file v1.2.1 // indirect
 	github.com/knadh/koanf/providers/rawbytes v1.0.0 // indirect
 	github.com/knadh/koanf/v2 v2.3.3 // indirect
+	github.com/linkedin/goavro/v2 v2.15.0 // indirect
 	github.com/lithammer/fuzzysearch v1.1.8 // indirect
 	github.com/mattn/go-runewidth v0.0.21 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect

go.sum

@@ -1727,6 +1727,8 @@ github.com/trivago/grok v1.0.0/go.mod h1:9t59xLInhrncYq9a3J7488NgiBZi5y5yC7bss+w
 github.com/trivago/tgo v1.0.7 h1:uaWH/XIy9aWYWpjm2CU3RpcqZXmX2ysQ9/Go+d9gyrM=
 github.com/trivago/tgo v1.0.7/go.mod h1:w4dpD+3tzNIIiIfkWWa85w5/B77tlvdZckQ+6PkFnhc=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
+github.com/twmb/avro v1.3.1 h1:OZw+qg/EhqW6rn0CrqzkcTiS33khJoSN08iPTo4Kubs=
+github.com/twmb/avro v1.3.1/go.mod h1:TUQS96Ptl8tDRyK0Jw91FXIVCoPKz4sXxvUSShiG5FA=
 github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY=
 github.com/twmb/franz-go v1.20.7/go.mod h1:0bRX9HZVaoueqFWhPZNi2ODnJL7DNa6mK0HeCrC2bNU=
 github.com/twmb/franz-go/pkg/kadm v1.17.2 h1:g5f1sAxnTkYC6G96pV5u715HWhxd66hWaDZUAQ8xHY8=

internal/impl/avro/processor.go

@@ -16,13 +16,14 @@ package avro
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"strings"
 
-	"github.com/linkedin/goavro/v2"
+	"github.com/twmb/avro"
 
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
@@ -65,16 +66,35 @@ func init() {
 
 type avroOperator func(part *service.Message) error
 
-func newAvroToJSONOperator(encoding string, codec *goavro.Codec) (avroOperator, error) {
+// setAvroStructured encodes a decoded Avro value as tagged-union JSON
+// and sets it as structured data on the message. The JSON round-trip
+// ensures all values are JSON-native types (float64, string, etc.)
+// for Bloblang compatibility.
+func setAvroStructured(schema *avro.Schema, native any, part *service.Message) error {
+	jb, err := schema.EncodeJSON(native, avro.TaggedUnions(), avro.TagLogicalTypes())
+	if err != nil {
+		return fmt.Errorf("converting Avro document to JSON: %v", err)
+	}
+	var jObj any
+	if err := json.Unmarshal(jb, &jObj); err != nil {
+		return fmt.Errorf("converting Avro document to JSON: %v", err)
+	}
+	part.SetStructuredMut(jObj)
+	return nil
+}
+
+func newAvroToJSONOperator(encoding string, schema *avro.Schema) (avroOperator, error) {
 	switch encoding {
 	case "textual":
+		// Input is Avro JSON bytes. Parse as standard JSON to preserve
+		// union wrapping ({"type": value} format) in the structured output.
 		return func(part *service.Message) error {
 			pBytes, err := part.AsBytes()
 			if err != nil {
 				return err
 			}
-			jObj, _, err := codec.NativeFromTextual(pBytes)
-			if err != nil {
+			var jObj any
+			if err := json.Unmarshal(pBytes, &jObj); err != nil {
 				return fmt.Errorf("converting Avro document to JSON: %v", err)
 			}
 			part.SetStructuredMut(jObj)
@@ -86,66 +106,83 @@ func newAvroToJSONOperator(encoding string, codec *goavro.Codec) (avroOperator,
 			if err != nil {
 				return err
 			}
-			jObj, _, err := codec.NativeFromBinary(pBytes)
-			if err != nil {
+			var native any
+			if _, err := schema.Decode(pBytes, &native); err != nil {
 				return fmt.Errorf("converting Avro document to JSON: %v", err)
 			}
-			part.SetStructuredMut(jObj)
-			return nil
+			return setAvroStructured(schema, native, part)
 		}, nil
 	case "single":
 		return func(part *service.Message) error {
 			pBytes, err := part.AsBytes()
 			if err != nil {
 				return err
 			}
-			jObj, _, err := codec.NativeFromSingle(pBytes)
-			if err != nil {
+			var native any
+			if _, err := schema.DecodeSingleObject(pBytes, &native); err != nil {
 				return fmt.Errorf("converting Avro document to JSON: %v", err)
 			}
-			part.SetStructuredMut(jObj)
-			return nil
+			return setAvroStructured(schema, native, part)
 		}, nil
 	}
 	return nil, fmt.Errorf("encoding '%v' not recognised", encoding)
 }
 
-func newAvroFromJSONOperator(encoding string, codec *goavro.Codec) (avroOperator, error) {
+func newAvroFromJSONOperator(encoding string, schema *avro.Schema) (avroOperator, error) {
 	switch encoding {
 	case "textual":
+		// Input is Avro JSON (with union wrapping). Decode it, then
+		// re-encode to Avro JSON to validate against the schema.
 		return func(part *service.Message) error {
-			jObj, err := part.AsStructured()
+			pBytes, err := part.AsBytes()
 			if err != nil {
-				return fmt.Errorf("parsing message as JSON: %v", err)
+				return fmt.Errorf("parsing message as bytes: %v", err)
+			}
+			// DecodeJSON unwraps unions; EncodeJSON re-wraps them.
+			var native any
+			if err := schema.DecodeJSON(pBytes, &native); err != nil {
+				return fmt.Errorf("converting JSON to Avro schema: %v", err)
 			}
-			var textual []byte
-			if textual, err = codec.TextualFromNative(nil, jObj); err != nil {
+			textual, err := schema.EncodeJSON(native, avro.TaggedUnions(), avro.TagLogicalTypes())
+			if err != nil {
 				return fmt.Errorf("converting JSON to Avro schema: %v", err)
 			}
 			part.SetBytes(textual)
 			return nil
 		}, nil
 	case "binary":
+		// Input is Avro JSON (with union wrapping). Decode it to
+		// native types, then encode to Avro binary.
 		return func(part *service.Message) error {
-			jObj, err := part.AsStructured()
+			pBytes, err := part.AsBytes()
 			if err != nil {
-				return fmt.Errorf("parsing message as JSON: %v", err)
+				return fmt.Errorf("parsing message as bytes: %v", err)
+			}
+			var native any
+			if err := schema.DecodeJSON(pBytes, &native); err != nil {
+				return fmt.Errorf("converting JSON to Avro schema: %v", err)
 			}
-			var binary []byte
-			if binary, err = codec.BinaryFromNative(nil, jObj); err != nil {
+			binary, err := schema.Encode(native)
+			if err != nil {
 				return fmt.Errorf("converting JSON to Avro schema: %v", err)
 			}
 			part.SetBytes(binary)
 			return nil
 		}, nil
 	case "single":
+		// Input is Avro JSON (with union wrapping). Decode it to
+		// native types, then encode to single-object format.
 		return func(part *service.Message) error {
-			jObj, err := part.AsStructured()
+			pBytes, err := part.AsBytes()
 			if err != nil {
-				return fmt.Errorf("parsing message as JSON: %v", err)
+				return fmt.Errorf("parsing message as bytes: %v", err)
 			}
-			var single []byte
-			if single, err = codec.SingleFromNative(nil, jObj); err != nil {
+			var native any
+			if err := schema.DecodeJSON(pBytes, &native); err != nil {
+				return fmt.Errorf("converting JSON to Avro schema: %v", err)
+			}
+			single, err := schema.AppendSingleObject(nil, native)
+			if err != nil {
 				return fmt.Errorf("converting JSON to Avro schema: %v", err)
 			}
 			part.SetBytes(single)
@@ -155,12 +192,12 @@ func newAvroFromJSONOperator(encoding string, codec *goavro.Codec) (avroOperator
 	return nil, fmt.Errorf("encoding '%v' not recognised", encoding)
 }
 
-func strToAvroOperator(opStr, encoding string, codec *goavro.Codec) (avroOperator, error) {
+func strToAvroOperator(opStr, encoding string, schema *avro.Schema) (avroOperator, error) {
 	switch opStr {
 	case "to_json":
-		return newAvroToJSONOperator(encoding, codec)
+		return newAvroToJSONOperator(encoding, schema)
 	case "from_json":
-		return newAvroFromJSONOperator(encoding, codec)
+		return newAvroFromJSONOperator(encoding, schema)
 	}
 	return nil, fmt.Errorf("operator not recognised: %v", opStr)
 }
@@ -192,13 +229,13 @@ func loadSchema(schemaPath string) (string, error) {
 
 //------------------------------------------------------------------------------
 
-type avro struct {
+type avroProcessor struct {
 	operator avroOperator
 	log      *service.Logger
 }
 
 func newAvroFromConfig(conf *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
-	a := &avro{log: mgr.Logger()}
+	a := &avroProcessor{log: mgr.Logger()}
 
 	var operator, encoding, schema, schemaPath string
 	var err error
@@ -227,20 +264,20 @@ func newAvroFromConfig(conf *service.ParsedConfig, mgr *service.Resources) (serv
 		return nil, errors.New("a schema must be specified with either the `schema` or `schema_path` fields")
 	}
 
-	codec, err := goavro.NewCodec(schema)
+	parsed, err := avro.Parse(schema)
 	if err != nil {
 		return nil, fmt.Errorf("parsing schema: %v", err)
 	}
 
-	if a.operator, err = strToAvroOperator(operator, encoding, codec); err != nil {
+	if a.operator, err = strToAvroOperator(operator, encoding, parsed); err != nil {
 		return nil, err
 	}
 	return a, nil
 }
 
 //------------------------------------------------------------------------------
 
-func (p *avro) Process(_ context.Context, msg *service.Message) (service.MessageBatch, error) {
+func (p *avroProcessor) Process(_ context.Context, msg *service.Message) (service.MessageBatch, error) {
 	err := p.operator(msg)
 	if err != nil {
 		p.log.Debugf("Operator failed: %v\n", err)
@@ -249,6 +286,6 @@ func (p *avro) Process(_ context.Context, msg *service.Message) (service.Message
 	return service.MessageBatch{msg}, nil
 }
 
-func (*avro) Close(context.Context) error {
+func (*avroProcessor) Close(context.Context) error {
 	return nil
 }

internal/impl/avro/scanner.go

@@ -15,12 +15,13 @@
 package avro
 
 import (
-	"bufio"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 
-	"github.com/linkedin/goavro/v2"
+	"github.com/twmb/avro"
+	"github.com/twmb/avro/ocf"
 
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
@@ -47,13 +48,13 @@ For example, the union schema ` + "`[\"null\",\"string\",\"Foo\"]`, where `Foo`"
 - the string ` + "`\"a\"` as `{\"string\": \"a\"}`" + `; and
 - a ` + "`Foo` instance as `{\"Foo\": {...}}`, where `{...}` indicates the JSON encoding of a `Foo`" + ` instance.
 
-However, it is possible to instead create documents in https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull[standard/raw JSON format^] by setting the field ` + "<<avro_raw_json,`avro_raw_json`>> to `true`" + `.
+However, it is possible to instead create documents in standard/raw JSON format by setting the field ` + "<<avro_raw_json,`avro_raw_json`>> to `true`" + `.
 
 This scanner also emits the canonical Avro schema as ` + "`@avro_schema`" + ` metadata, along with the schema's fingerprint available via ` + "`@avro_schema_fingerprint`" + `.
 `).
 		Fields(
 			service.NewBoolField(sFieldRawJSON).
-				Description("Whether 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.").
+				Description("Whether messages should be decoded into normal JSON rather than https://avro.apache.org/docs/current/specification/_print/#json-encoding[Avro JSON^]. When true, union values are unwrapped (bare values instead of {\"type\": value} wrappers).").
 				Advanced().
 				Default(false),
 		)
@@ -79,24 +80,22 @@ type avroScannerCreator struct {
 }
 
 func (c *avroScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, _ *service.ScannerSourceDetails) (service.BatchScanner, error) {
-	br := bufio.NewReader(rdr)
-	ocf, err := goavro.NewOCFReader(br)
+	reader, err := ocf.NewReader(rdr)
 	if err != nil {
 		return nil, err
 	}
 
-	ocfCodec := ocf.Codec()
-	ocfSchema := ocfCodec.Schema()
-	if c.rawJSON {
-		if ocfCodec, err = goavro.NewCodecForStandardJSONFull(ocfSchema); err != nil {
-			return nil, err
-		}
-	}
+	schema := reader.Schema()
+	canonical := string(schema.Canonical())
+	fp := schema.Fingerprint(avro.NewRabin())
 
 	return service.AutoAggregateBatchScannerAcks(&avroScanner{
 		r:         rdr,
-		ocf:       ocf,
-		avroCodec: ocfCodec,
+		reader:    reader,
+		schema:    schema,
+		rawJSON:   c.rawJSON,
+		canonical: canonical,
+		fp:        fp,
 	}, aFn), nil
 }
 
@@ -106,41 +105,48 @@ func (*avroScannerCreator) Close(context.Context) error {
 
 type avroScanner struct {
 	r         io.ReadCloser
-	ocf       *goavro.OCFReader
-	avroCodec *goavro.Codec
+	reader    *ocf.Reader
+	schema    *avro.Schema
+	rawJSON   bool
+	canonical string
+	fp        []byte
 }
 
 func (c *avroScanner) NextBatch(context.Context) (service.MessageBatch, error) {
 	if c.r == nil {
 		return nil, io.EOF
 	}
 
-	if !c.ocf.Scan() {
-		err := c.ocf.Err()
-		if err != nil {
-			return nil, fmt.Errorf("scanning OCF file: %s", err)
+	var native any
+	if err := c.reader.Decode(&native); err != nil {
+		if err == io.EOF {
+			return nil, io.EOF
 		}
-		return nil, io.EOF
-	}
-
-	datum, err := c.ocf.Read()
-	if err != nil {
 		return nil, fmt.Errorf("reading OCF datum: %s", err)
 	}
 
-	jb, err := c.avroCodec.TextualFromNative(nil, datum)
+	var jb []byte
+	var err error
+	if c.rawJSON {
+		jb, err = c.schema.EncodeJSON(native)
+	} else {
+		jb, err = c.schema.EncodeJSON(native, avro.TaggedUnions(), avro.TagLogicalTypes())
+	}
 	if err != nil {
-		return nil, fmt.Errorf("decoding OCF datum to JSON: %s", err)
+		return nil, fmt.Errorf("encoding OCF datum to JSON: %s", err)
 	}
+
 	msg := service.NewMessage(jb)
-	msg.MetaSetMut("avro_schema", c.avroCodec.CanonicalSchema())
-	msg.MetaSetMut("avro_schema_fingerprint", c.avroCodec.Rabin)
+	msg.MetaSetMut("avro_schema", c.canonical)
+	msg.MetaSetMut("avro_schema_fingerprint", c.fp)
 	return service.MessageBatch{msg}, nil
 }
 
 func (c *avroScanner) Close(context.Context) error {
 	if c.r == nil {
 		return nil
 	}
-	return c.r.Close()
+	readerErr := c.reader.Close() // closes codec only, not the underlying reader
+	rErr := c.r.Close()
+	return errors.Join(readerErr, rErr)
 }