This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 10749a27f35 Release subscription WAL buffers between prefetch rounds
(#18956)
10749a27f35 is described below
commit 10749a27f354022c5d0b9779912db1a0efae3b4c
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 16:05:30 2026 +0800
Release subscription WAL buffers between prefetch rounds (#18956)
* Release subscription WAL read buffers between prefetch rounds
* Stream Prometheus snapshots in bounded UTF-8 chunks
* Handle shortened WAL segments after buffer release
---
.../dataregion/wal/io/ProgressWALReader.java | 4 ++
.../dataregion/wal/io/WALByteBufReader.java | 4 ++
.../dataregion/wal/io/WALInputStream.java | 45 ++++++++++++-
.../consensus/ConsensusPrefetchingQueue.java | 10 +++
.../broker/consensus/ProgressWALIterator.java | 9 +++
.../wal/compression/WALCompressionTest.java | 73 +++++++++++++++++++++
.../dataregion/wal/io/ProgressWALReaderTest.java | 75 ++++++++++++++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 3 +
.../reporter/prometheus/PrometheusReporter.java | 26 +++++++-
.../prometheus/PrometheusReporterTest.java | 57 ++++++++++++++++
10 files changed, 303 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReader.java
index 9b6c67e0716..bac5c17c45f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReader.java
@@ -78,6 +78,10 @@ public class ProgressWALReader implements Closeable {
return delegate.getCurrentEntryIndex();
}
+ public void releaseReadBuffers() {
+ delegate.releaseReadBuffers();
+ }
+
@Override
public void close() throws IOException {
delegate.close();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
index f89b1fb5393..cd4e9c2ce50 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
@@ -132,6 +132,10 @@ public class WALByteBufReader implements Closeable {
return metaData;
}
+ public void releaseReadBuffers() {
+ walInputStream.releaseReadBuffers();
+ }
+
@Override
public void close() throws IOException {
logStream.close();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
index 1c20982be67..d4b4a8b78f9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
@@ -52,6 +52,10 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
private final ByteBuffer compressedSizeBuffer =
ByteBuffer.allocate(Integer.BYTES);
private ByteBuffer dataBuffer = null;
private ByteBuffer compressedBuffer = null;
+ private long currentSegmentOffset = -1;
+ private long suspendedSegmentOffset = -1;
+ private int suspendedSegmentPosition;
+ private int suspendedSegmentRemaining;
private final long fileSize;
File logFile;
/*
@@ -205,7 +209,26 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
@Override
public void close() throws IOException {
- channel.close();
+ try {
+ channel.close();
+ } finally {
+ releaseReadBuffers();
+ suspendedSegmentOffset = -1;
+ suspendedSegmentRemaining = 0;
+ }
+ }
+
+ /**
+ * Releases segment scratch buffers while preserving the unread position. A
paused reader must not
+ * pin a complete WAL segment for each subscription queue. The next read
reloads only the current
+ * segment, without reopening the file or rescanning entry metadata.
+ */
+ public void releaseReadBuffers() {
+ if (dataBuffer != null && dataBuffer.hasRemaining()) {
+ suspendedSegmentOffset = currentSegmentOffset;
+ suspendedSegmentPosition = dataBuffer.position();
+ suspendedSegmentRemaining = dataBuffer.remaining();
+ }
MmapUtil.clean(dataBuffer);
MmapUtil.clean(compressedBuffer);
dataBuffer = null;
@@ -217,11 +240,16 @@ public class WALInputStream extends InputStream
implements AutoCloseable {
long size = (endOffset - channel.position());
if (!Objects.isNull(dataBuffer)) {
size += dataBuffer.limit() - dataBuffer.position();
+ } else {
+ size += suspendedSegmentRemaining;
}
return (int) size;
}
private void loadNextSegment() throws IOException {
+ if (suspendedSegmentOffset >= 0) {
+ channel.position(suspendedSegmentOffset);
+ }
if (channel.position() >= endOffset) {
throw new
EOFException(StorageEngineMessages.REACH_END_OFFSET_OF_WAL_FILE);
}
@@ -234,6 +262,15 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
} else {
tryLoadSegment();
}
+ currentSegmentOffset = startPosition;
+ if (suspendedSegmentOffset >= 0) {
+ if (suspendedSegmentPosition >= dataBuffer.limit()) {
+ throw new EOFException(StorageEngineMessages.UNEXPECTED_END_OF_FILE);
+ }
+ dataBuffer.position(suspendedSegmentPosition);
+ suspendedSegmentOffset = -1;
+ suspendedSegmentRemaining = 0;
+ }
WritingMetrics.getInstance()
.recordWALRead(channel.position() - startPosition, System.nanoTime() -
startTime);
}
@@ -349,6 +386,9 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
* @throws IOException If the file is broken or the given position is invalid
*/
public void skipToGivenLogicalPosition(long pos) throws IOException {
+ releaseReadBuffers();
+ suspendedSegmentOffset = -1;
+ suspendedSegmentRemaining = 0;
if (version == WALFileVersion.V2 || version == WALFileVersion.V3) {
channel.position(version.getVersionBytes().length);
long posRemain = pos;
@@ -360,6 +400,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
posRemain -= segmentInfo.uncompressedSize;
channel.position(currentPos + segmentInfo.dataInDiskSize +
segmentInfo.headerSize());
} else {
+ currentSegmentOffset = currentPos;
break;
}
} while (posRemain >= 0);
@@ -389,7 +430,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
public void read(ByteBuffer buffer) throws IOException {
int totalBytesToBeRead = buffer.remaining();
while (totalBytesToBeRead > 0) {
- if (dataBuffer.remaining() == 0) {
+ if (dataBuffer == null || dataBuffer.remaining() == 0) {
loadNextSegment();
}
int currReadBytes = Math.min(dataBuffer.remaining(), totalBytesToBeRead);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index 812fc0f6f7d..3f0561d37f6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -1948,6 +1948,16 @@ public class ConsensusPrefetchingQueue {
} finally {
final ProgressWALIterator iterator = subscriptionWALIterator;
walReplayState = Objects.nonNull(iterator) ? iterator.toString() :
"UNAVAILABLE";
+ // Readers outlive a round, but their direct segment buffers must not be
retained by every
+ // inactive/backpressured queue. Keep their cursors and request
fragments for the next round.
+ if (Objects.nonNull(iterator)) {
+ iterator.releaseReadBuffers();
+ }
+ for (final WriterReplayState state : orderedWriterReplay.values()) {
+ if (Objects.nonNull(state.iterator)) {
+ state.iterator.releaseReadBuffers();
+ }
+ }
releaseReadLock();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
index 4b1a58a3009..07438858a35 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
@@ -562,6 +562,15 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
resetCurrentFileTracking();
}
+ /**
+ * Drops segment buffers between prefetch rounds without discarding buffered
request fragments.
+ */
+ public void releaseReadBuffers() {
+ if (currentReader != null) {
+ currentReader.releaseReadBuffers();
+ }
+ }
+
private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws
IOException {
while (true) {
if (scanBudget != null && scanBudget.shouldYield()) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
index abbc766deb9..07c317c4bff 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
@@ -53,10 +53,13 @@ import org.junit.Test;
import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
+import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
@@ -113,6 +116,76 @@ public class WALCompressionTest {
testSkipToGivenPosition();
}
+ @Test
+ public void testReleaseReadBuffersPreservesPositionAcrossSegmentsAndSeeks()
throws Exception {
+ byte[] expected = new byte[24];
+ for (int i = 0; i < expected.length; i++) {
+ expected[i] = (byte) i;
+ }
+ for (CompressionType compression :
+ new CompressionType[] {CompressionType.UNCOMPRESSED,
CompressionType.LZ4}) {
+
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(compression);
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+ writer.setCompressedByteBuffer(ByteBuffer.allocate(128));
+ for (int start = 0; start < expected.length; start += 8) {
+ ByteBuffer segment = ByteBuffer.allocate(8);
+ segment.put(expected, start, 8);
+ writer.write(segment);
+ }
+ }
+ try (WALInputStream stream = new WALInputStream(walFile)) {
+ Assert.assertEquals(0, stream.read());
+ int available = stream.available();
+ stream.releaseReadBuffers();
+ stream.releaseReadBuffers();
+ Assert.assertEquals(available, stream.available());
+ Assert.assertEquals(1, stream.getLogicalReadOffset());
+ byte[] crossSegment = new byte[10];
+ stream.read(crossSegment);
+ Assert.assertArrayEquals(java.util.Arrays.copyOfRange(expected, 1,
11), crossSegment);
+ stream.releaseReadBuffers();
+ stream.skipToGivenLogicalPosition(4);
+ stream.releaseReadBuffers();
+ ByteBuffer restOfSegment = ByteBuffer.allocate(4);
+ stream.read(restOfSegment);
+ Assert.assertArrayEquals(
+ java.util.Arrays.copyOfRange(expected, 4, 8),
restOfSegment.array());
+ // A pause exactly at the segment boundary must load the next segment
without replaying it.
+ stream.releaseReadBuffers();
+ Assert.assertEquals(8, stream.read());
+ stream.releaseReadBuffers();
+ ByteBuffer tail = ByteBuffer.allocate(15);
+ stream.read(tail);
+ Assert.assertArrayEquals(java.util.Arrays.copyOfRange(expected, 9,
24), tail.array());
+ }
+ Files.delete(walFile.toPath());
+ }
+ }
+
+ @Test
+ public void testReleasedReadBufferDetectsShortenedSegment() throws Exception
{
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+ ByteBuffer segment = ByteBuffer.allocate(8);
+ segment.putLong(42L);
+ writer.write(segment);
+ }
+ try (WALInputStream stream = new WALInputStream(walFile)) {
+ for (int i = 0; i < 3; i++) {
+ stream.read();
+ }
+ stream.releaseReadBuffers();
+ try (FileChannel channel = FileChannel.open(walFile.toPath(),
StandardOpenOption.WRITE)) {
+ channel.position(WALFileVersion.V3.getVersionBytes().length);
+ ByteBuffer header = ByteBuffer.allocate(Byte.BYTES + Integer.BYTES);
+ header.put(CompressionType.UNCOMPRESSED.serialize()).putInt(2).flip();
+ while (header.hasRemaining()) {
+ channel.write(header);
+ }
+ }
+ Assert.assertThrows(EOFException.class, stream::read);
+ }
+ }
+
@Test
public void testSkipToGivenPositionWithoutCompression()
throws NoSuchFieldException,
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReaderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReaderTest.java
index 755964e1d25..29a6dfdd56c 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReaderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReaderTest.java
@@ -19,19 +19,94 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.dataregion.wal.WALTestUtils;
+
+import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.junit.Test;
import java.io.File;
+import java.lang.management.BufferPoolMXBean;
+import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ProgressWALReaderTest {
+ @Test
+ public void testPausedReadersReleaseDirectMemoryAndResumeWithinSegment()
throws Exception {
+ Path dir = Files.createTempDirectory("progress-wal-reader-buffers");
+ File walFile = dir.resolve("test.wal").toFile();
+ CompressionType originalCompression =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+ long originalMinCompressionSize = WALTestUtils.getMinCompressionSize();
+ List<ProgressWALReader> readers = new ArrayList<>();
+ try {
+ WALTestUtils.setMinCompressionSize(0);
+ for (CompressionType compression :
+ new CompressionType[] {CompressionType.UNCOMPRESSED,
CompressionType.LZ4}) {
+
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(compression);
+ final int entrySize = 1024 * 1024;
+ byte[] first = new byte[entrySize];
+ byte[] second = new byte[entrySize];
+ Arrays.fill(first, (byte) 11);
+ Arrays.fill(second, (byte) 22);
+ ByteBuffer segment = ByteBuffer.allocate(entrySize * 2);
+ segment.put(first).put(second);
+ WALMetaData metadata = new WALMetaData();
+ metadata.add(entrySize, 10, 1, 1000, 1, 10);
+ metadata.add(entrySize, 11, 1, 2000, 1, 11);
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+
writer.setCompressedByteBuffer(ByteBuffer.allocate(segment.capacity() * 2));
+ writer.write(segment, metadata);
+ }
+ BufferPoolMXBean directPool =
+
ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class).stream()
+ .filter(pool -> "direct".equals(pool.getName()))
+ .findFirst()
+ .get();
+ long initialDirectMemory = directPool.getMemoryUsed();
+ for (int i = 0; i < 32; i++) {
+ ProgressWALReader reader = new ProgressWALReader(walFile);
+ readers.add(reader);
+ assertArrayEquals(first, reader.next().array());
+ reader.releaseReadBuffers();
+ reader.releaseReadBuffers();
+ }
+ // All readers remain open halfway through the same segment. Their
retained direct memory
+ // must not grow with the number of subscriptions (64 MiB without
releasing the buffers).
+ assertTrue(directPool.getMemoryUsed() - initialDirectMemory <
entrySize);
+ for (ProgressWALReader reader : readers) {
+ assertArrayEquals(second, reader.next().array());
+ assertEquals(1, reader.getCurrentEntryIndex());
+ assertEquals(11, reader.getCurrentEntryLocalSeq());
+ assertFalse(reader.hasNext());
+ reader.releaseReadBuffers();
+ reader.close();
+ }
+ readers.clear();
+ Files.delete(walFile.toPath());
+ }
+ } finally {
+ for (ProgressWALReader reader : readers) {
+ reader.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(originalCompression);
+ WALTestUtils.setMinCompressionSize(originalMinCompressionSize);
+ Files.deleteIfExists(walFile.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testReadWriterProgressMetadataFromV3Wal() throws Exception {
Path dir = Files.createTempDirectory("progress-wal-reader");
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
index a4bfb29304d..5202b0ef1b7 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
@@ -94,6 +94,9 @@ public class ProgressWALIteratorTest {
assertTrue(iterator.hasYieldedScan());
assertFalse(iterator.hasBufferedNext());
assertFalse(iterator.hasIncompleteScan());
+ iterator.releaseReadBuffers();
+ // Releasing the segment must preserve both the physical cursor and
request fragments.
+ iterator.releaseReadBuffers();
// Neither a skipped prefix nor an unfinished follower request is a
completed delivery.
assertFalse(iterator.hasNext(budget));
}
diff --git
a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java
b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java
index 00ae6c6b988..30ce9573637 100644
---
a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java
+++
b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java
@@ -46,6 +46,7 @@ import io.netty.handler.ssl.SslContextBuilder;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;
@@ -77,6 +78,7 @@ public class PrometheusReporter implements Reporter {
private static final MetricConfig METRIC_CONFIG =
MetricConfigDescriptor.getInstance().getMetricConfig();
private static final long PROMETHEUS_DEFAULT_SCRAPE_INTERVAL_SECONDS = 15;
+ private static final int PROMETHEUS_RESPONSE_CHUNK_SIZE = 8 * 1024;
private final AbstractMetricManager metricManager;
private final Supplier<ScheduledExecutorService>
snapshotUpdateExecutorSupplier;
private volatile ScheduledExecutorService snapshotUpdateExecutor;
@@ -148,7 +150,8 @@ public class PrometheusReporter implements Reporter {
? getMetricsSnapshot()
: scrape();
return res.header(HttpHeaderNames.CONTENT_TYPE,
"text/plain")
- .sendString(Mono.just(metrics));
+ .sendString(
+ splitMetricsIntoChunks(metrics),
StandardCharsets.UTF_8);
}));
if (METRIC_CONFIG.isEnableSSL()) {
SslContext sslContext;
@@ -189,6 +192,27 @@ public class PrometheusReporter implements Reporter {
return true;
}
+ static Flux<String> splitMetricsIntoChunks(final String metrics) {
+ // Encoding one snapshot into a single direct ByteBuf requires a
contiguous allocation sized
+ // for every series. Generate bounded chunks on demand so slow scrapes do
not pin that buffer.
+ return Flux.generate(
+ () -> 0,
+ (start, sink) -> {
+ if (start >= metrics.length()) {
+ sink.complete();
+ return start;
+ }
+ int end = start + Math.min(PROMETHEUS_RESPONSE_CHUNK_SIZE,
metrics.length() - start);
+ if (end < metrics.length()
+ && Character.isHighSurrogate(metrics.charAt(end - 1))
+ && Character.isLowSurrogate(metrics.charAt(end))) {
+ end--;
+ }
+ sink.next(metrics.substring(start, end));
+ return end;
+ });
+ }
+
@SuppressWarnings("unsafeThreadSchedule")
private void startSnapshotUpdater() {
// Keep metric collection off Reactor HTTP threads and avoid overlapping
scrapes.
diff --git
a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java
b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java
index 7072777ac66..5b19d7f5f31 100644
---
a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java
+++
b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java
@@ -24,7 +24,13 @@ import
org.apache.iotdb.metrics.config.MetricConfigDescriptor;
import org.apache.iotdb.metrics.impl.DoNothingMetricManager;
import org.junit.Test;
+import reactor.netty.DisposableServer;
+import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
@@ -72,4 +78,55 @@ public class PrometheusReporterTest {
executors.forEach(ScheduledExecutorService::shutdownNow);
}
}
+
+ @Test
+ public void testMetricsResponseChunksPreserveContentAndSurrogatePairs() {
+ String metrics = "x".repeat(8 * 1024 - 1) + "\uD83D\uDE00" + "y".repeat(8
* 1024 + 1);
+
+ List<String> chunks =
PrometheusReporter.splitMetricsIntoChunks(metrics).collectList().block();
+
+ assertEquals(metrics, String.join("", chunks));
+ assertTrue(chunks.stream().allMatch(chunk -> chunk.length() <= 8 * 1024));
+ assertEquals(8 * 1024 - 1, chunks.get(0).length());
+ assertTrue(chunks.get(1).startsWith("\uD83D\uDE00"));
+ }
+
+ @Test
+ public void testLargeMetricsSnapshotCanBeScraped() throws Exception {
+ MetricConfig metricConfig =
MetricConfigDescriptor.getInstance().getMetricConfig();
+ boolean originalAsyncUpdate =
metricConfig.isPrometheusReporterAsyncUpdate();
+ Integer originalPort = metricConfig.getPrometheusReporterPort();
+ metricConfig.setPrometheusReporterAsyncUpdate(true);
+ metricConfig.setPrometheusReporterPort(0);
+ PrometheusReporter reporter = new PrometheusReporter(new
DoNothingMetricManager());
+ HttpURLConnection connection = null;
+ try {
+ assertTrue(reporter.start());
+ // Also run this regression with -XX:MaxDirectMemorySize=32m: a single
response buffer for
+ // this snapshot cannot fit, whereas streaming it requires only bounded
direct memory.
+ String metrics = "x".repeat(8 * 1024 - 1) + "\uD83D\uDE00" +
"y".repeat(24 * 1024 * 1024);
+ Field snapshotField =
PrometheusReporter.class.getDeclaredField("metricsSnapshot");
+ snapshotField.setAccessible(true);
+ snapshotField.set(reporter, metrics);
+ Field serverField =
PrometheusReporter.class.getDeclaredField("httpServer");
+ serverField.setAccessible(true);
+ int port = ((DisposableServer) serverField.get(reporter)).port();
+ connection =
+ (HttpURLConnection)
+ URI.create("http://127.0.0.1:"; + port +
"/metrics").toURL().openConnection();
+ connection.setConnectTimeout(5_000);
+ connection.setReadTimeout(10_000);
+ assertEquals(200, connection.getResponseCode());
+ try (InputStream response = connection.getInputStream()) {
+ assertEquals(metrics, new String(response.readAllBytes(),
StandardCharsets.UTF_8));
+ }
+ } finally {
+ if (connection != null) {
+ connection.disconnect();
+ }
+ reporter.stop();
+ metricConfig.setPrometheusReporterAsyncUpdate(originalAsyncUpdate);
+ metricConfig.setPrometheusReporterPort(originalPort);
+ }
+ }
}
