[improve] Use single buffer for metrics when noUnsafe use (apache#23612)

Co-authored-by: Lari Hotari <[email protected]>
(cherry picked from commit 9b79d70)
(cherry picked from commit aee413b)

build/run_unit_group.sh

@@ -77,6 +77,8 @@ alias echo='{ [[ $- =~ .*x.* ]] && trace_enabled=1 || trace_enabled=0; set +x; }
 # Test Groups  -- start --
 function test_group_broker_group_1() {
   mvn_test -pl pulsar-broker -Dgroups='broker' -DtestReuseFork=true
+  # run tests in broker-isolated group individually (instead of with -Dgroups=broker-isolated) to avoid scanning all test classes
+  mvn_test -pl pulsar-broker -Dtest=org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGeneratorWithNoUnsafeTest -DtestForkCount=1 -DtestReuseFork=false
 }
 
 function test_group_broker_group_2() {

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;
@@ -363,19 +364,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/PrometheusMetricsGeneratorWithNoUnsafeTest.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.stats.prometheus;
+
+import static org.testng.Assert.assertFalse;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.internal.PlatformDependent;
+import java.time.Clock;
+import lombok.Cleanup;
+import org.apache.pulsar.common.util.SimpleTextOutputStream;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-isolated")
+public class PrometheusMetricsGeneratorWithNoUnsafeTest {
+
+    @BeforeClass
+    static void setup() {
+        System.setProperty("io.netty.noUnsafe", "true");
+    }
+
+    @Test
+    public void testWriteStringWithNoUnsafe() {
+        assertFalse(PlatformDependent.hasUnsafe());
+        @Cleanup
+        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);
+        outputStream.write("test");
+    }
+}