From e910de5f6ec289e48761f6a096f7c6c45a0db7c2 Mon Sep 17 00:00:00 2001 From: yuzone Date: Sat, 28 Mar 2026 00:50:48 +0900 Subject: [PATCH 1/3] fix: guard type assertion in getFLBPluginContext with ok pattern FLBPluginGetContext returns interface{}. The previous code used an unguarded type assertion .(int) which panics if the stored value is not an int (e.g. when FlushCtx is called for an instance whose Init never completed successfully). Replace with the two-value form and log a warning on mismatch, returning 0 so the subsequent configMap lookup fails gracefully. --- out_writeapi.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/out_writeapi.go b/out_writeapi.go index b0a3902..8065c48 100644 --- a/out_writeapi.go +++ b/out_writeapi.go @@ -602,7 +602,12 @@ var newDecoder = func(data unsafe.Pointer, length int) *output.FLBDecoder { // Mock it whenever needed var getFLBPluginContext = func(ctx unsafe.Pointer) int { if ctx != nil { - return output.FLBPluginGetContext(ctx).(int) + id, ok := output.FLBPluginGetContext(ctx).(int) + if !ok { + log.Printf("FLBPluginGetContext returned unexpected type for context pointer %p", ctx) + return 0 + } + return id } return 0 } From 62125f408db7be78e8b8c1c0a30cffa7a56a9106 Mon Sep 17 00:00:00 2001 From: yuzone Date: Sat, 28 Mar 2026 00:51:18 +0900 Subject: [PATCH 2/3] fix: guard nil dereference in getOffset (test-only helper) configMap[id] returns nil when the id is not registered. The previous code dereferenced config unconditionally, causing a panic. Add ok-pattern check on the map lookup and a bounds check on the stream slice before accessing index 0. --- out_writeapi.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/out_writeapi.go b/out_writeapi.go index 8065c48..9c0ab4d 100644 --- a/out_writeapi.go +++ b/out_writeapi.go @@ -491,8 +491,14 @@ func getLeastLoadedStream(streamSlice *[]*streamConfig) int { // This is a test-only method which takes in a config id and returns the current offset value of the struct corresponding to the id func getOffset(id int) int64 { - config := configMap[id] + config, ok := configMap[id] + if !ok || config == nil { + return 0 + } streamSlice := *config.managedStreamSlice + if len(streamSlice) == 0 { + return 0 + } return streamSlice[0].offsetCounter } From f3adeeec05f8d0bc00d73b0ded439a4322b9473a Mon Sep 17 00:00:00 2001 From: yuzone Date: Sat, 28 Mar 2026 01:02:00 +0900 Subject: [PATCH 3/3] fix: release messagePool entries on FLBPluginExitCtx to prevent unbounded growth messagePool is a global sync.Map keyed by protoreflect.MessageDescriptor pointer identity. Entries are created lazily on the first Flush call for each descriptor and were never removed, so descriptor objects (and their associated *dynamicpb.Message caches) could not be GC'd after a config was torn down. Add cleanupMsgPool() which traverses the full descriptor tree rooted at a given MessageDescriptor and calls sync.Map.Delete for each node, including nested STRUCT sub-message descriptors. Call it at the end of FLBPluginExitCtx so that each config's pool entries are released when the output instance is shut down. Add TestCleanupMsgPool covering flat and nested schemas to verify that both the top-level and sub-message pool entries are removed after cleanup. --- map_to_proto.go | 23 ++++++++++++++++ map_to_proto_test.go | 62 ++++++++++++++++++++++++++++++++++++++++++++ out_writeapi.go | 5 ++++ 3 files changed, 90 insertions(+) diff --git a/map_to_proto.go b/map_to_proto.go index 72f1f59..35c501d 100644 --- a/map_to_proto.go +++ b/map_to_proto.go @@ -89,6 +89,29 @@ func getPooledMessage(md protoreflect.MessageDescriptor) *dynamicpb.Message { return p.(*sync.Pool).Get().(*dynamicpb.Message) } +// cleanupMsgPool removes all messagePool entries for the descriptor tree +// rooted at md. Call this when an outputConfig is torn down to prevent +// pool entries (and their associated descriptor references) from accumulating +// across init/exit cycles. +func cleanupMsgPool(md protoreflect.MessageDescriptor) { + cleanupMsgPoolRecursive(md, make(map[protoreflect.MessageDescriptor]bool)) +} + +func cleanupMsgPoolRecursive(md protoreflect.MessageDescriptor, visited map[protoreflect.MessageDescriptor]bool) { + if visited[md] { + return + } + visited[md] = true + messagePool.Delete(md) + fields := md.Fields() + for i := 0; i < fields.Len(); i++ { + f := fields.Get(i) + if f.Kind() == protoreflect.MessageKind || f.Kind() == protoreflect.GroupKind { + cleanupMsgPoolRecursive(f.Message(), visited) + } + } +} + // putPooledMessage clears all populated fields and returns the message to the pool. // Clearing via Range+Clear preserves internal map bucket memory so that // subsequent reuse avoids re-growing the map. diff --git a/map_to_proto_test.go b/map_to_proto_test.go index 4fbfb37..f6c292d 100644 --- a/map_to_proto_test.go +++ b/map_to_proto_test.go @@ -1141,6 +1141,68 @@ func TestMessagePoolReuse_RawPath_NestedRecord(t *testing.T) { assert.Equal(t, int64(0), msg2.Get(md.Fields().ByName("nested")).Message().Get(nestedMd.Fields().ByName("y")).Int()) } +// TestCleanupMsgPool verifies that cleanupMsgPool removes all messagePool +// entries for a flat schema and a nested (STRUCT) schema. +func TestCleanupMsgPool(t *testing.T) { + t.Run("flat schema", func(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "name", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "count", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + cache := buildFieldLookupCache(md) + + // Populate the pool by serialising one record. + _, err := mapToBinary(md, map[string]interface{}{"name": "x", "count": int64(1)}, cache) + require.NoError(t, err) + + _, populated := messagePool.Load(md) + assert.True(t, populated, "messagePool should have entry before cleanup") + + cleanupMsgPool(md) + + _, populated = messagePool.Load(md) + assert.False(t, populated, "messagePool should have no entry after cleanup") + }) + + t.Run("nested schema cleans sub-message entries", func(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + { + Name: "nested", + Type: storagepb.TableFieldSchema_STRUCT, + Mode: storagepb.TableFieldSchema_NULLABLE, + Fields: []*storagepb.TableFieldSchema{ + {Name: "x", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + }, + }, + } + md := buildMD(t, schema) + nestedMd := md.Fields().ByName("nested").Message() + cache := buildFieldLookupCache(md) + + _, err := rawMapToBinary(md, map[interface{}]interface{}{ + "nested": map[interface{}]interface{}{"x": []byte("hello")}, + }, cache) + require.NoError(t, err) + + _, topPopulated := messagePool.Load(md) + _, subPopulated := messagePool.Load(nestedMd) + assert.True(t, topPopulated, "top-level entry should exist before cleanup") + assert.True(t, subPopulated, "sub-message entry should exist before cleanup") + + cleanupMsgPool(md) + + _, topPopulated = messagePool.Load(md) + _, subPopulated = messagePool.Load(nestedMd) + assert.False(t, topPopulated, "top-level entry should be removed after cleanup") + assert.False(t, subPopulated, "sub-message entry should be removed after cleanup") + }) +} + // BenchmarkRawMapToBinary benchmarks the raw map[interface{}]interface{} path. func BenchmarkRawMapToBinary(b *testing.B) { schema := mangleInputSchema(&storagepb.TableSchema{ diff --git a/out_writeapi.go b/out_writeapi.go index 9c0ab4d..4d3df50 100644 --- a/out_writeapi.go +++ b/out_writeapi.go @@ -906,6 +906,11 @@ func FLBPluginExitCtx(ctx unsafe.Pointer) int { if errFlag { return output.FLB_ERROR } + + // Release messagePool entries for this config's descriptor tree so that + // the descriptors and their cached *dynamicpb.Message objects can be GC'd. + cleanupMsgPool(config.messageDescriptor) + return output.FLB_OK }