RSocketRequesterTracingObservationHandler producing netty buffer LEAKs

[kevinat]

I'm getting some warnings about memory leaks with Spring boot 3.2, and after searching I found that according to:
Netty reports LEAK while using reactor (Rsocket) with Sleuth #2256
there seems to be a bug fix that hasn't migrated.

Here is @OlegDokuka 's fix migrated to RSocketRequesterTracingObservationHandler:

--- a/rsocket-micrometer/src/main/java/io/rsocket/micrometer/observation/RSocketRequesterTracingObservationHandler.java
+++ b/rsocket-micrometer/src/main/java/io/rsocket/micrometer/observation/RSocketRequesterTracingObservationHandler.java
@@ -23,13 +23,17 @@ import io.micrometer.tracing.Tracer;
 import io.micrometer.tracing.handler.TracingObservationHandler;
 import io.micrometer.tracing.internal.EncodingUtils;
 import io.micrometer.tracing.propagation.Propagator;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
 import io.rsocket.Payload;
+import io.rsocket.metadata.CompositeMetadataCodec;
 import io.rsocket.metadata.TracingMetadataCodec;
-import java.util.HashSet;
+import io.rsocket.metadata.WellKnownMimeType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashSet;
+
 public class RSocketRequesterTracingObservationHandler
 implements TracingObservationHandler<RSocketContext> {
 private static final Logger log =
@@ -115,17 +119,25 @@ public class RSocketRequesterTracingObservationHandler
 long[] spanId = EncodingUtils.fromString(traceContext.spanId());
 long[] parentSpanId = EncodingUtils.fromString(traceContext.parentId());
 boolean isTraceId128Bit = traceIds.length == 2;
+ final ByteBufAllocator allocator = newMetadata.alloc();
 if (isTraceId128Bit) {
- TracingMetadataCodec.encode128(
- newMetadata.alloc(),
- traceIds[0],
- traceIds[1],
- spanId[0],
- EncodingUtils.fromString(traceContext.parentId())[0],
- flags);
+ CompositeMetadataCodec.encodeAndAddMetadata(newMetadata,
+ allocator, WellKnownMimeType.MESSAGE_RSOCKET_TRACING_ZIPKIN,
+ TracingMetadataCodec.encode128(
+ allocator,
+ traceIds[0],
+ traceIds[1],
+ spanId[0],
+ EncodingUtils.fromString(traceContext.parentId())[0],
+ flags));
 } else {
- TracingMetadataCodec.encode64(
- newMetadata.alloc(), traceIds[0], spanId[0], parentSpanId[0], flags);
+ CompositeMetadataCodec.encodeAndAddMetadata(newMetadata,
+ allocator, WellKnownMimeType.MESSAGE_RSOCKET_TRACING_ZIPKIN,
+ TracingMetadataCodec.encode64(
+ allocator, traceIds[0], spanId[0], parentSpanId[0], flags));
 }
 }
 }