[fix][broker] Avoid OOM not trigger PulsarByteBufAllocator outOfMemor…

…yListener when use ByteBufAllocator.DEFAULT.heapBuffer in PrometheusMetricsGeneratorUtils (#18747) Co-authored-by: nicklixinyang <nicklixinyang@didiglobal.com>
apache · Dec 8, 2022 · b268847 · b268847
1 parent 68ca60c
commit b268847
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 6 deletions.
diff --git a/.../main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGeneratorUtils.java b/.../main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGeneratorUtils.java
@@ -19,14 +19,14 @@
 package org.apache.pulsar.broker.stats.prometheus;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.prometheus.client.Collector;
 import io.prometheus.client.CollectorRegistry;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.Enumeration;
 import java.util.List;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
 
 /**
@@ -38,7 +38,7 @@ public class PrometheusMetricsGeneratorUtils {
     public static void generate(String cluster, OutputStream out,
                                 List<PrometheusRawMetricsProvider> metricsProviders)
             throws IOException {
-        ByteBuf buf = ByteBufAllocator.DEFAULT.heapBuffer();
+        ByteBuf buf = PulsarByteBufAllocator.DEFAULT.heapBuffer();
         try {
             SimpleTextOutputStream stream = new SimpleTextOutputStream(buf);
             generateSystemMetrics(stream, cluster);

diff --git a/...r/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java b/...r/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -22,7 +22,6 @@
 import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGeneratorUtils.getTypeStr;
 import static org.apache.pulsar.common.stats.JvmMetrics.getJvmDirectMemoryUsed;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import io.prometheus.client.Collector;
 import io.prometheus.client.CollectorRegistry;
@@ -51,6 +50,7 @@
 import org.apache.pulsar.broker.stats.metrics.ManagedCursorMetrics;
 import org.apache.pulsar.broker.stats.metrics.ManagedLedgerCacheMetrics;
 import org.apache.pulsar.broker.stats.metrics.ManagedLedgerMetrics;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.stats.Metrics;
 import org.apache.pulsar.common.util.DirectMemoryUtils;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
@@ -138,7 +138,7 @@ public static synchronized void generate(PulsarService pulsar, boolean includeTo
                 } catch (IOException e) {
                     log.error("Generate metrics failed", e);
                     //return empty buffer if exception happens
-                    return ByteBufAllocator.DEFAULT.heapBuffer(0);
+                    return PulsarByteBufAllocator.DEFAULT.heapBuffer(0);
                 }
             });
 

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java b/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java
@@ -31,7 +31,6 @@
 import com.google.gson.JsonSerializationContext;
 import com.google.gson.JsonSerializer;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufUtil;
 import java.io.IOException;
 import java.lang.reflect.Type;
@@ -44,6 +43,7 @@
 import java.util.SortedMap;
 import java.util.TreeMap;
 import org.apache.pulsar.client.internal.DefaultImplementation;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
@@ -367,7 +367,7 @@ public static byte[] convertKeyValueDataStringToSchemaInfoSchema(
         int dataLength = 4 + keyBytes.length + 4 + valueBytes.length;
         byte[] schema = new byte[dataLength];
         //record the key value schema respective length
-        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.heapBuffer(dataLength);
+        ByteBuf byteBuf = PulsarByteBufAllocator.DEFAULT.heapBuffer(dataLength);
         byteBuf.writeInt(keyBytes.length).writeBytes(keyBytes).writeInt(valueBytes.length).writeBytes(valueBytes);
         byteBuf.readBytes(schema);
         return schema;