Skip to content

Fixing goavro bug due to codec state mutation #4070

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yaron2 merged 3 commits into from
Oct 22, 2025
Merged

Fixing goavro bug due to codec state mutation #4070

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5dafa9b
Working around current goavro bug (https://github.com/linkedin/goavro…
passuied Oct 22, 2025
efe40a6
fixing flaky test (unrelated to my PR)
passuied Oct 22, 2025
8563cd0
Upgrade goavro to 2.14.1 to get rid of mutation state issue.
passuied Oct 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 27 additions & 3 deletions common/component/kafka/kafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,9 @@ type Kafka struct {
latestSchemaCacheWriteLock sync.RWMutex
latestSchemaCacheReadLock sync.Mutex

// Whether to encode/decode Avro into Avro JSON or standard JSON
useAvroJSON bool

// used for background logic that cannot use the context passed to the Init function
internalContext context.Context
internalContextCancel func()
Expand Down Expand Up @@ -228,6 +231,7 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
k.srClient = srclient.CreateSchemaRegistryClient(meta.SchemaRegistryURL)
k.srClient.CodecCreationEnabled(true)
k.srClient.CodecJsonEnabled(!meta.UseAvroJSON)
k.useAvroJSON = meta.UseAvroJSON
// Empty password is a possibility
if meta.SchemaRegistryAPIKey != "" {
k.srClient.SetCredentials(meta.SchemaRegistryAPIKey, meta.SchemaRegistryAPISecret)
Expand Down Expand Up @@ -405,18 +409,27 @@ func (k *Kafka) getLatestSchema(topic string) (*srclient.Schema, *goavro.Codec,
if errSchema != nil {
return nil, nil, errSchema
}
codec := schema.Codec()

codec, errCodec := k.getCodec(schema)
if errCodec != nil {
return nil, nil, errCodec
}
defer k.latestSchemaCacheWriteLock.Unlock()
k.latestSchemaCacheWriteLock.Lock()
k.latestSchemaCache[subject] = SchemaCacheEntry{schema: schema, codec: codec, expirationTime: time.Now().Add(k.latestSchemaCacheTTL)}
k.latestSchemaCacheWriteLock.Unlock()

return schema, codec, nil
}
schema, err := srClient.GetLatestSchema(getSchemaSubject(topic))
if err != nil {
return nil, nil, err
}
return schema, schema.Codec(), nil
codec, errCodec := k.getCodec(schema)
if errCodec != nil {
return nil, nil, errCodec
}

return schema, codec, nil
}

func (k *Kafka) getSchemaRegistyClient() (srclient.ISchemaRegistryClient, error) {
Expand All @@ -427,6 +440,17 @@ func (k *Kafka) getSchemaRegistyClient() (srclient.ISchemaRegistryClient, error)
return k.srClient, nil
}

func (k *Kafka) getCodec(schema *srclient.Schema) (*goavro.Codec, error) {
// The data coming through is either Avro JSON or standard JSON.
// Force creation of a new codec instance for serialization and deserialization to avoid state mutation issues.
// https://github.com/linkedin/goavro/issues/299
// Once the bug is fixed, we can remove this and use the codec directly from schema.Codec()
if k.useAvroJSON {
return goavro.NewCodec(schema.Schema())
}
return goavro.NewCodecForStandardJSONFull(schema.Schema())
}

func (k *Kafka) SerializeValue(topic string, data []byte, metadata map[string]string) ([]byte, error) {
// Null Data is valid and a tombstone record.
// It should be converted to NULL and not go through schema validation & encoding
Expand Down
4 changes: 4 additions & 0 deletions common/component/kafka/kafka_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ func TestDeserializeValue(t *testing.T) {
logger: logger.NewLogger("kafka_test"),
}
kJSON.srClient.CodecJsonEnabled(true)
kJSON.useAvroJSON = false
schemaJSON, _ := registryJSON.CreateSchema("my-topic-value", testSchema1, srclient.Avro)

// set up for Standard JSON
Expand Down Expand Up @@ -249,12 +250,14 @@ func TestSerializeValueCachingDisabled(t *testing.T) {
srClient: registryJSON,
schemaCachingEnabled: false,
logger: logger.NewLogger("kafka_test"),
useAvroJSON: false,
}

kAvroJSON := Kafka{
srClient: registryAvroJSON,
schemaCachingEnabled: false,
logger: logger.NewLogger("kafka_test"),
useAvroJSON: true,
}

t.Run("valueSchemaType not set, leave value as is", func(t *testing.T) {
Expand Down Expand Up @@ -327,6 +330,7 @@ func TestSerializeValueCachingEnabled(t *testing.T) {
latestSchemaCache: make(map[string]SchemaCacheEntry),
latestSchemaCacheTTL: time.Minute * 5,
logger: logger.NewLogger("kafka_test"),
useAvroJSON: false,
}

t.Run("valueSchemaType not set, leave value as is", func(t *testing.T) {
Expand Down
28 changes: 21 additions & 7 deletions middleware/http/wasm/internal/e2e_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,15 +80,23 @@ func Test_EndToEnd(t *testing.T) {
// init (main) and the request[0-9] funcs to info level.
//
// Then, we expect to see stdout and stderr from both scopes
// at debug level.
// at debug level. Allow duplicates from multi-module pools by
// checking substrings instead of exact combined lines.
for _, s := range []string{
`level=info msg="main ConsoleLog"`,
`level=info msg="request[0] ConsoleLog"`,
`level=debug msg="wasm stdout: main Stdout\nrequest[0] Stdout\n"`,
`level=debug msg="wasm stderr: main Stderr\nrequest[0] Stderr\n"`,
} {
require.Contains(t, log.String(), s)
}

// stdout
require.Contains(t, log.String(), `level=debug msg="wasm stdout:`)
require.Contains(t, log.String(), "main Stdout")
require.Contains(t, log.String(), "request[0] Stdout")
// stderr
require.Contains(t, log.String(), `level=debug msg="wasm stderr:`)
require.Contains(t, log.String(), "main Stderr")
require.Contains(t, log.String(), "request[0] Stderr")
},
},
{
Expand All @@ -108,14 +116,20 @@ func Test_EndToEnd(t *testing.T) {
for _, s := range []string{
`level=info msg="main ConsoleLog"`,
`level=info msg="request[0] ConsoleLog"`,
`level=debug msg="wasm stdout: main Stdout\nrequest[0] Stdout\n"`,
`level=debug msg="wasm stderr: main Stderr\nrequest[0] Stderr\n"`,
`level=info msg="request[1] ConsoleLog"`,
`level=debug msg="wasm stdout: request[1] Stdout\n"`,
`level=debug msg="wasm stderr: request[1] Stderr\n"`,
} {
require.Contains(t, log.String(), s)
}
// Allow duplicates for main/request[0] stdout/stderr across modules.
require.Contains(t, log.String(), `level=debug msg="wasm stdout:`)
require.Contains(t, log.String(), "main Stdout")
require.Contains(t, log.String(), "request[0] Stdout")
require.Contains(t, log.String(), `level=debug msg="wasm stderr:`)
require.Contains(t, log.String(), "main Stderr")
require.Contains(t, log.String(), "request[0] Stderr")
// And ensure request[1] appears in stdout/stderr logs too.
require.Contains(t, log.String(), "request[1] Stdout")
require.Contains(t, log.String(), "request[1] Stderr")
},
},
{
Expand Down