[improve] Use single buffer for metrics when noUnsafe use

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java

@@ -25,6 +25,7 @@
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.buffer.Unpooled;
+import io.netty.util.internal.PlatformDependent;
 import io.prometheus.client.Collector;
 import java.io.BufferedOutputStream;
 import java.io.IOException;
@@ -364,19 +365,24 @@ protected ByteBuf generateMetrics(List<PrometheusRawMetricsProvider> metricsProv
         }
     }
 
-    private ByteBuf allocateMultipartCompositeDirectBuffer() {
+    ByteBuf allocateMultipartCompositeDirectBuffer() {
         // use composite buffer with pre-allocated buffers to ensure that the pooled allocator can be used
         // for allocating the buffers
         ByteBufAllocator byteBufAllocator = PulsarByteBufAllocator.DEFAULT;
-        int chunkSize = resolveChunkSize(byteBufAllocator);
-        CompositeByteBuf buf = byteBufAllocator.compositeDirectBuffer(
+        ByteBuf buf;
+        if (PlatformDependent.hasUnsafe()) {
+            int chunkSize = resolveChunkSize(byteBufAllocator);
+            buf = byteBufAllocator.compositeDirectBuffer(
                 Math.max(MINIMUM_FOR_MAX_COMPONENTS, (initialBufferSize / chunkSize) + 1));
-        int totalLen = 0;
-        while (totalLen < initialBufferSize) {
-            totalLen += chunkSize;
-            // increase the capacity in increments of chunkSize to preallocate the buffers
-            // in the composite buffer
-            buf.capacity(totalLen);
+            int totalLen = 0;
+            while (totalLen < initialBufferSize) {
+                totalLen += chunkSize;
+                // increase the capacity in increments of chunkSize to preallocate the buffers
+                // in the composite buffer
+                buf.capacity(totalLen);
+            }
+        } else {
+            buf = byteBufAllocator.directBuffer(initialBufferSize);
         }
         return buf;
     }

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGeneratorTest.java

@@ -32,13 +32,22 @@
 import java.util.Random;
 import java.util.concurrent.ExecutionException;
 import java.util.zip.GZIPInputStream;
+import lombok.Cleanup;
 import org.apache.commons.io.IOUtils;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.SimpleTextOutputStream;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 public class PrometheusMetricsGeneratorTest {
 
+
+    @BeforeClass
+    public static void setProperties() {
+        System.setProperty("io.netty.noUnsafe", "true");
+    }
+
     // reproduce issue #22575
     @Test
     public void testReproducingBufferOverflowExceptionAndEOFExceptionBugsInGzipCompression()
@@ -82,4 +91,22 @@ protected ByteBuf generateMetrics(List<PrometheusRawMetricsProvider> metricsProv
             metricsBuffer.release();
         }
     }
-}
+
+    @Test
+    public void testWriteStringWithNoUnsafe() {
+        PrometheusMetricsGenerator generator = new PrometheusMetricsGenerator(null, false, false, false, false,
+                Clock.systemUTC());
+        @Cleanup("release")
+        ByteBuf buf = generator.allocateMultipartCompositeDirectBuffer();
+        for (int i = 0; i < 2; i++) {
+            buf.writeBytes(new byte[1024 * 1024]);
+        }
+
+        SimpleTextOutputStream outputStream = new SimpleTextOutputStream(buf);
+        try {
+            outputStream.write("trest");
+        } catch (Exception e) {
+
+        }
+    }
+}