[otlp] Fix ProtobufOtlpLogSerializer pool handling (#5989)

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpLogSerializer.cs

@@ -1,6 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+using System.Diagnostics;
 using OpenTelemetry.Internal;
 using OpenTelemetry.Logs;
 using OpenTelemetry.Trace;
@@ -34,6 +35,15 @@ internal static int WriteLogsData(byte[] buffer, int writePosition, SdkLimitOpti
                 ScopeLogsList[scopeName] = logRecords;
             }
 
+            if (logRecord.Source == LogRecord.LogRecordSource.FromSharedPool)
+            {
+                Debug.Assert(logRecord.PoolReferenceCount > 0, "logRecord PoolReferenceCount value was unexpected");
+
+                // Note: AddReference call here prevents the LogRecord from
+                // being given back to the pool by Batch<LogRecord>.
+                logRecord.AddReference();
+            }
+
             logRecords.Add(logRecord);
         }
 
@@ -50,6 +60,18 @@ internal static void ReturnLogRecordListToPool()
         {
             foreach (var entry in ScopeLogsList)
             {
+                foreach (var logRecord in entry.Value)
+                {
+                    if (logRecord.Source == LogRecord.LogRecordSource.FromSharedPool)
+                    {
+                        Debug.Assert(logRecord.PoolReferenceCount > 0, "logRecord PoolReferenceCount value was unexpected");
+
+                        // Note: Try to return the LogRecord to the shared pool
+                        // now that work is done.
+                        LogRecordSharedPool.Current.Return(logRecord);
+                    }
+                }
+
                 entry.Value.Clear();
                 LogsListPool.Push(entry.Value);
             }