You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/09/11 22:37:05 UTC
(pulsar) branch branch-3.0 updated (9b27ba2c9b3 -> bd2ec48bbba)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 9b27ba2c9b3 [fix][broker]Fix never recovered metadata store bad version issue if received a large response from ZK (#24580)
new 26a31ac8a5c [improve][build] Disable javadoc build failure (#24594)
new cb5a3abbb94 [fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
new 498571defe2 [fix][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads (#24725)
new bd2ec48bbba [fix][io] Improve Kafka Connect source offset flushing logic (#24654)
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 2 +
.../metrics/DataSketchesOpStatsLogger.java | 87 +---------
.../metrics/DataSketchesSummaryLogger.java | 49 +-----
.../prometheus/metrics/ThreadLocalAccessor.java | 134 +++++++++++++++
.../metrics/ThreadLocalAccessorTest.java | 92 ++++++++++
.../PersistentAcknowledgmentsGroupingTracker.java | 29 ++--
.../impl/AcknowledgementsGroupingTrackerTest.java | 30 ++--
pulsar-io/kafka-connect-adaptor/pom.xml | 6 +
.../kafka/connect/AbstractKafkaConnectSource.java | 141 +++++++++-------
.../io/kafka/connect/KafkaConnectSourceTest.java | 186 ++++++++++++++++++++-
10 files changed, 534 insertions(+), 222 deletions(-)
create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
(pulsar) 04/04: [fix][io] Improve Kafka Connect source offset flushing logic (#24654)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bd2ec48bbba3a185755d6b75b36ecc1979e5527e
Author: Harangozó Péter <ha...@gmail.com>
AuthorDate: Thu Sep 11 23:18:12 2025 +0200
[fix][io] Improve Kafka Connect source offset flushing logic (#24654)
(cherry picked from commit a824f04a36e168e415284d599fbe2fb94caa58c8)
---
pulsar-io/kafka-connect-adaptor/pom.xml | 6 +
.../kafka/connect/AbstractKafkaConnectSource.java | 141 +++++++++-------
.../io/kafka/connect/KafkaConnectSourceTest.java | 186 ++++++++++++++++++++-
3 files changed, 270 insertions(+), 63 deletions(-)
diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml
index 76b9c7ebd3c..679c3be089a 100644
--- a/pulsar-io/kafka-connect-adaptor/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -179,6 +179,12 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-broker</artifactId>
diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java
index 270decd8ab5..e048ddf7244 100644
--- a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java
+++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java
@@ -31,11 +31,13 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.source.SourceRecord;
@@ -68,7 +70,6 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
// pulsar io related variables
private Iterator<SourceRecord> currentBatch = null;
- private CompletableFuture<Void> flushFuture;
private OffsetBackingStore offsetStore;
private OffsetStorageReader offsetReader;
private String topicNamespace;
@@ -76,6 +77,8 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
public OffsetStorageWriter offsetWriter;
// number of outstandingRecords that have been polled but not been acked
private final AtomicInteger outstandingRecords = new AtomicInteger(0);
+ private final AtomicBoolean flushing = new AtomicBoolean(false);
+ private final AtomicReference<CompletableFuture<Void>> flushFutureRef = new AtomicReference<>();
public static final String CONNECTOR_CLASS = "kafkaConnectorSourceClass";
@@ -162,38 +165,103 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
} else {
return processRecord;
}
} else {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well, then do next round read.
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
try {
- flushFuture.get();
+ if (snapshotFlushFuture != null) {
+ snapshotFlushFuture.get();
+ }
} catch (ExecutionException ex) {
- // log the error, continue execution
log.error("execution exception while get flushFuture", ex);
throw new Exception("Flush failed", ex.getCause());
} finally {
- flushFuture = null;
+ flushing.set(false);
+ // Clear only if this is still the current batch future
+ flushFutureRef.compareAndSet(snapshotFlushFuture, null);
currentBatch = null;
}
}
@@ -272,62 +340,19 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
return this.value == null;
}
- private void completedFlushOffset(Throwable error, Void result) {
- if (error != null) {
- log.error("Failed to flush offsets to storage: ", error);
- currentBatch = null;
- offsetWriter.cancelFlush();
- flushFuture.completeExceptionally(new Exception("No Offsets Added Error"));
- } else {
- try {
- sourceTask.commit();
-
- log.info("Finished flushing offsets to storage");
- currentBatch = null;
- flushFuture.complete(null);
- } catch (InterruptedException exception) {
- log.warn("Flush of {} offsets interrupted, cancelling", this);
- Thread.currentThread().interrupt();
- offsetWriter.cancelFlush();
- flushFuture.completeExceptionally(new Exception("Failed to commit offsets", exception));
- } catch (Throwable t) {
- // SourceTask can throw unchecked ConnectException/KafkaException.
- // Make sure the future is cancelled in that case
- log.warn("Flush of {} offsets failed, cancelling", this);
- offsetWriter.cancelFlush();
- flushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
- }
- }
- }
-
@Override
public void ack() {
- // TODO: should flush for each batch. not wait for a time for acked all.
- // How to handle order between each batch. QueueList<pair<batch, automicInt>>. check if head is all acked.
- boolean canFlush = (outstandingRecords.decrementAndGet() == 0);
-
- // consumed all the records, flush the offsets
- if (canFlush && flushFuture != null) {
- if (!offsetWriter.beginFlush()) {
- log.error("When beginFlush, No offsets to commit!");
- flushFuture.completeExceptionally(new Exception("No Offsets Added Error when beginFlush"));
- return;
- }
-
- Future<Void> doFlush = offsetWriter.doFlush(this::completedFlushOffset);
- if (doFlush == null) {
- // Offsets added in processSourceRecord, But here no offsets to commit
- log.error("No offsets to commit!");
- flushFuture.completeExceptionally(new Exception("No Offsets Added Error"));
- return;
- }
+ // Decrement and let the centralized flusher decide if we should flush now
+ if (outstandingRecords.decrementAndGet() == 0) {
+ triggerOffsetsFlushIfNeeded();
}
}
@Override
public void fail() {
- if (flushFuture != null) {
- flushFuture.completeExceptionally(new Exception("Sink Error"));
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ if (snapshotFlushFuture != null) {
+ snapshotFlushFuture.completeExceptionally(new Exception("Sink Error"));
}
}
}
diff --git a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
index 7347d07ee0e..54449da4e1f 100644
--- a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
+++ b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.io.kafka.connect;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
@@ -28,19 +30,25 @@ import static org.testng.Assert.assertEquals;
import java.io.File;
import java.io.OutputStream;
import java.nio.file.Files;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.kafka.connect.file.FileStreamSourceConnector;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.storage.OffsetStorageWriter;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.SourceContext;
import org.apache.pulsar.io.kafka.connect.KafkaConnectSource.KafkaSourceRecord;
+import org.awaitility.Awaitility;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -49,7 +57,7 @@ import org.testng.annotations.Test;
* Test the implementation of {@link KafkaConnectSource}.
*/
@Slf4j
-public class KafkaConnectSourceTest extends ProducerConsumerBase {
+public class KafkaConnectSourceTest extends ProducerConsumerBase {
private String offsetTopicName;
// The topic to publish data to, for kafkaSource
@@ -96,6 +104,174 @@ public class KafkaConnectSourceTest extends ProducerConsumerBase {
testOpenAndReadTask(config);
}
+ private OffsetStorageWriter wrapOffsetWriterWithSpy(KafkaConnectSource source) throws IllegalAccessException {
+ OffsetStorageWriter original = source.getOffsetWriter();
+ OffsetStorageWriter spy = org.mockito.Mockito.spy(original);
+ FieldUtils.writeField(source, "offsetWriter", spy, true);
+ return spy;
+ }
+
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter spyWriter = wrapOffsetWriterWithSpy(kafkaConnectSource);
+
+ try (OutputStream os = Files.newOutputStream(tempFile.toPath())) {
+ os.write("first\n".getBytes());
+ os.flush();
+ os.write("second\n".getBytes());
+ os.flush();
+ }
+
+ Thread t = new Thread(() -> {
+ try {
+ kafkaConnectSource.read();
+ } catch (Exception e) {
+ // Ignore, it just needs the loop running through a batch where every message is filtered out
+ log.error("Exception in read thread", e);
+ }
+ });
+ t.setDaemon(true);
+ t.start();
+
+ try {
+ // First ensure offsets are being written for filtered records.
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(10))
+ .untilAsserted(() -> {
+ Mockito.verify(spyWriter, Mockito.atLeastOnce())
+ .offset(ArgumentMatchers.any(), ArgumentMatchers.any());
+ });
+
+ // Then ensure a flush cycle is triggered without waiting for any acks (since all are filtered).
+ Awaitility.await()
+ .atMost(java.time.Duration.ofSeconds(10))
+ .untilAsserted(() -> {
+ Mockito.verify(spyWriter, Mockito.atLeastOnce())
+ .beginFlush();
+ Mockito.verify(spyWriter, Mockito.atLeastOnce())
+ .doFlush(ArgumentMatchers.any());
+ });
+ } finally {
+ kafkaConnectSource.close();
+ t.interrupt();
+ }
+ }
+
+ @Test(timeOut = 40000)
+ public void testNoFlushUntilAck() throws Exception {
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter spyWriter = wrapOffsetWriterWithSpy(kafkaConnectSource);
+
+ try (OutputStream os = Files.newOutputStream(tempFile.toPath())) {
+ os.write("one\n".getBytes());
+ os.flush();
+ }
+
+ Record<KeyValue<byte[], byte[]>> readRecord = kafkaConnectSource.read();
+
+ // Verify no flush happens while there is one outstanding record (no ack yet)
+ verify(spyWriter, timeout(10000).times(0)).beginFlush();
+ verify(spyWriter, timeout(10000).times(0))
+ .doFlush(ArgumentMatchers.any());
+
+ readRecord.ack();
+
+ // Verify flush happens after ack
+ verify(spyWriter, timeout(10000).atLeastOnce()).beginFlush();
+ verify(spyWriter, timeout(10000).atLeastOnce())
+ .doFlush(ArgumentMatchers.any());
+
+ kafkaConnectSource.close();
+ }
+
+ @Test(timeOut = 40000)
+ public void testPartialAckNoFlush_ThenFlushOnAllAck() throws Exception {
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter spyWriter = wrapOffsetWriterWithSpy(kafkaConnectSource);
+
+ try (OutputStream os = Files.newOutputStream(tempFile.toPath())) {
+ os.write("first\n".getBytes());
+ os.flush();
+ os.write("second\n".getBytes());
+ os.flush();
+ }
+
+ Record<KeyValue<byte[], byte[]>> r1 = kafkaConnectSource.read();
+ Record<KeyValue<byte[], byte[]>> r2 = kafkaConnectSource.read();
+
+ // Ack only the first; one outstanding remains -> no flush should happen
+ r1.ack();
+ verify(spyWriter, timeout(10000).times(0)).beginFlush();
+ verify(spyWriter, timeout(10000).times(0))
+ .doFlush(ArgumentMatchers.any());
+
+ // Ack the second; outstanding reaches zero -> flush should be triggered
+ r2.ack();
+ verify(spyWriter, timeout(10000).atLeastOnce()).beginFlush();
+ verify(spyWriter, timeout(10000).atLeastOnce())
+ .doFlush(ArgumentMatchers.any());
+
+ kafkaConnectSource.close();
+ }
+
+ @Test(timeOut = 20000)
+ public void testAckFlush() throws Exception {
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter spyWriter = wrapOffsetWriterWithSpy(kafkaConnectSource);
+
+ try (OutputStream os = Files.newOutputStream(tempFile.toPath())) {
+ os.write("first\n".getBytes());
+ os.flush();
+ os.write("second\n".getBytes());
+ os.flush();
+ }
+
+ Record<KeyValue<byte[], byte[]>> r1 = kafkaConnectSource.read();
+ Record<KeyValue<byte[], byte[]>> r2 = kafkaConnectSource.read();
+
+ // Ack both, no outstanding remains -> flush should happen
+ r1.ack();
+ r2.ack();
+
+ verify(spyWriter, timeout(10000).atLeastOnce()).beginFlush();
+ verify(spyWriter, timeout(10000).atLeastOnce())
+ .doFlush(ArgumentMatchers.any());
+
+ kafkaConnectSource.close();
+ }
+
+
@Test
public void testOpenAndReadTaskDirect() throws Exception {
Map<String, Object> config = getConfig();
@@ -156,8 +332,8 @@ public class KafkaConnectSourceTest extends ProducerConsumerBase {
sourceOffset.put("test", 0);
value.put("myField", "42");
SourceRecord srcRecord = new SourceRecord(
- sourcePartition, sourceOffset, topicName, null,
- null, null, null, value
+ sourcePartition, sourceOffset, topicName, null,
+ null, null, null, value
);
KafkaSourceRecord record = kafkaConnectSource.processSourceRecord(srcRecord);
@@ -196,8 +372,8 @@ public class KafkaConnectSourceTest extends ProducerConsumerBase {
Map<String, Object> value = new HashMap<>();
value.put("myField", "42");
SourceRecord record = new SourceRecord(
- null, null, "test-topic", null,
- null, null, null, value
+ null, null, "test-topic", null,
+ null, null, null, value
);
SourceRecord transformed = kafkaConnectSource.applyTransforms(record);
(pulsar) 03/04: [fix][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads (#24725)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 498571defe29afaf85f924e3c8badbc66c6174c9
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Sep 11 22:19:35 2025 +0800
[fix][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads (#24725)
(cherry picked from commit 14543d3fde935d1b70e3707a8b2c0294eb53dccb)
---
.../PersistentAcknowledgmentsGroupingTracker.java | 29 +++++++++++----------
.../impl/AcknowledgementsGroupingTrackerTest.java | 30 ++++++++--------------
2 files changed, 25 insertions(+), 34 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
index d30c3de0fd7..5e462d9e25a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.impl;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
+import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.FastThreadLocal;
@@ -26,13 +27,12 @@ import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -82,7 +82,8 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
* broker.
*/
private final ConcurrentSkipListSet<MessageIdAdv> pendingIndividualAcks;
- private final ConcurrentHashMap<MessageIdAdv, ConcurrentBitSetRecyclable> pendingIndividualBatchIndexAcks;
+ @VisibleForTesting
+ final ConcurrentSkipListMap<MessageIdAdv, ConcurrentBitSetRecyclable> pendingIndividualBatchIndexAcks;
private final ScheduledFuture<?> scheduledTask;
private final boolean batchIndexAckEnabled;
@@ -92,7 +93,7 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
EventLoopGroup eventLoopGroup) {
this.consumer = consumer;
this.pendingIndividualAcks = new ConcurrentSkipListSet<>();
- this.pendingIndividualBatchIndexAcks = new ConcurrentHashMap<>();
+ this.pendingIndividualBatchIndexAcks = new ConcurrentSkipListMap<>();
this.acknowledgementGroupTimeMicros = conf.getAcknowledgementsGroupTimeMicros();
this.maxAckGroupSize = conf.getMaxAcknowledgmentGroupSize();
this.batchIndexAckEnabled = conf.isBatchIndexAckEnabled();
@@ -324,7 +325,8 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
}
}
- private CompletableFuture<Void> doIndividualBatchAckAsync(MessageIdAdv msgId) {
+ @VisibleForTesting
+ CompletableFuture<Void> doIndividualBatchAckAsync(MessageIdAdv msgId) {
ConcurrentBitSetRecyclable bitSet = pendingIndividualBatchIndexAcks.computeIfAbsent(
MessageIdAdvUtils.discardBatch(msgId), __ -> {
final BitSet ackSet = msgId.getAckSet();
@@ -484,16 +486,15 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
}
}
- if (!pendingIndividualBatchIndexAcks.isEmpty()) {
- Iterator<Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable>> iterator =
- pendingIndividualBatchIndexAcks.entrySet().iterator();
-
- while (iterator.hasNext()) {
- Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry = iterator.next();
- entriesToAck.add(Triple.of(
- entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
- iterator.remove();
+ while (true) {
+ Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry =
+ pendingIndividualBatchIndexAcks.pollFirstEntry();
+ if (entry == null) {
+ // The entry has been removed in a different thread
+ break;
}
+ entriesToAck.add(Triple.of(
+ entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
}
if (entriesToAck.size() > 0) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
index efa398afba6..ebcc99f8557 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
@@ -31,23 +31,19 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
import java.util.BitSet;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.pulsar.client.api.MessageId;
+import lombok.var;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.util.TimedCompletableFuture;
import org.apache.pulsar.common.api.proto.CommandAck.AckType;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.common.api.proto.ProtocolVersion;
-import org.apache.pulsar.common.util.collections.ConcurrentBitSetRecyclable;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -388,24 +384,18 @@ public class AcknowledgementsGroupingTrackerTest {
}
@Test
- public void testDoIndividualBatchAckAsync() throws Exception{
+ public void testDoIndividualBatchAckAsync() {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
- AcknowledgmentsGroupingTracker tracker = new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
- MessageId messageId1 = new BatchMessageIdImpl(5, 1, 0, 3, 10, null);
+ var tracker = new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ var messageId1 = new BatchMessageIdImpl(5, 1, 0, 3, 10, null);
BitSet bitSet = new BitSet(20);
- for(int i = 0; i < 20; i ++) {
+ for (int i = 0; i < 20; i++) {
bitSet.set(i, true);
}
- MessageId messageId2 = new BatchMessageIdImpl(3, 2, 0, 5, 20, bitSet);
- Method doIndividualBatchAckAsync = PersistentAcknowledgmentsGroupingTracker.class
- .getDeclaredMethod("doIndividualBatchAckAsync", MessageIdAdv.class);
- doIndividualBatchAckAsync.setAccessible(true);
- doIndividualBatchAckAsync.invoke(tracker, messageId1);
- doIndividualBatchAckAsync.invoke(tracker, messageId2);
- Field pendingIndividualBatchIndexAcks = PersistentAcknowledgmentsGroupingTracker.class.getDeclaredField("pendingIndividualBatchIndexAcks");
- pendingIndividualBatchIndexAcks.setAccessible(true);
- ConcurrentHashMap<MessageIdAdv, ConcurrentBitSetRecyclable> batchIndexAcks =
- (ConcurrentHashMap<MessageIdAdv, ConcurrentBitSetRecyclable>) pendingIndividualBatchIndexAcks.get(tracker);
+ var messageId2 = new BatchMessageIdImpl(3, 2, 0, 5, 20, bitSet);
+ tracker.doIndividualBatchAckAsync(messageId1);
+ tracker.doIndividualBatchAckAsync(messageId2);
+ var batchIndexAcks = tracker.pendingIndividualBatchIndexAcks;
MessageIdImpl position1 = new MessageIdImpl(5, 1, 0);
MessageIdImpl position2 = new MessageIdImpl(3, 2, 0);
assertTrue(batchIndexAcks.containsKey(position1));
(pulsar) 02/04: [fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cb5a3abbb94a0c5918f41471bfd42a3226c02fe7
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Sep 10 00:12:36 2025 +0800
[fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
(cherry picked from commit 97156533ee0afb05f0538fbe4e5cae379c71b95a)
---
.../metrics/DataSketchesOpStatsLogger.java | 87 ++-----------
.../metrics/DataSketchesSummaryLogger.java | 49 +-------
.../prometheus/metrics/ThreadLocalAccessor.java | 134 +++++++++++++++++++++
.../metrics/ThreadLocalAccessorTest.java | 92 ++++++++++++++
4 files changed, 237 insertions(+), 125 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
index 12f54ba48a4..8973ba6a25c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
@@ -19,15 +19,10 @@
package org.apache.pulsar.broker.stats.prometheus.metrics;
import com.yahoo.sketches.quantiles.DoublesSketch;
-import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
import com.yahoo.sketches.quantiles.DoublesUnion;
import com.yahoo.sketches.quantiles.DoublesUnionBuilder;
-import io.netty.util.concurrent.FastThreadLocal;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
-import java.util.concurrent.locks.StampedLock;
import org.apache.bookkeeper.stats.OpStatsData;
import org.apache.bookkeeper.stats.OpStatsLogger;
@@ -65,15 +60,7 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
failCountAdder.increment();
failSumAdder.add((long) valueMillis);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.failSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateFail(valueMillis);
}
@Override
@@ -82,45 +69,21 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
successCountAdder.increment();
successSumAdder.add((long) valueMillis);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(valueMillis);
}
@Override
public void registerSuccessfulValue(long value) {
successCountAdder.increment();
successSumAdder.add(value);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(value);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(value);
}
@Override
public void registerFailedValue(long value) {
failCountAdder.increment();
failSumAdder.add(value);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.failSketch.update(value);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateFail(value);
}
@Override
@@ -141,21 +104,11 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
current = replacement;
replacement = local;
- final DoublesUnion aggregateSuccesss = new DoublesUnionBuilder().build();
+ final DoublesUnion aggregateSuccess = new DoublesUnionBuilder().build();
final DoublesUnion aggregateFail = new DoublesUnionBuilder().build();
- local.map.forEach((localData, b) -> {
- long stamp = localData.lock.writeLock();
- try {
- aggregateSuccesss.update(localData.successSketch);
- localData.successSketch.reset();
- aggregateFail.update(localData.failSketch);
- localData.failSketch.reset();
- } finally {
- localData.lock.unlockWrite(stamp);
- }
- });
-
- successResult = aggregateSuccesss.getResultAndReset();
+ local.record(aggregateSuccess, aggregateFail);
+
+ successResult = aggregateSuccess.getResultAndReset();
failResult = aggregateFail.getResultAndReset();
}
@@ -171,28 +124,4 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
DoublesSketch s = success ? successResult : failResult;
return s != null ? s.getQuantile(quantile) : Double.NaN;
}
-
- private static class LocalData {
- private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
- private final DoublesSketch failSketch = new DoublesSketchBuilder().build();
- private final StampedLock lock = new StampedLock();
- }
-
- private static class ThreadLocalAccessor {
- private final Map<LocalData, Boolean> map = new ConcurrentHashMap<>();
- private final FastThreadLocal<LocalData> localData = new FastThreadLocal<LocalData>() {
-
- @Override
- protected LocalData initialValue() throws Exception {
- LocalData localData = new LocalData();
- map.put(localData, Boolean.TRUE);
- return localData;
- }
-
- @Override
- protected void onRemoval(LocalData value) throws Exception {
- map.remove(value);
- }
- };
- }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
index cc144e0eb69..42c189d4bf3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
@@ -19,15 +19,10 @@
package org.apache.pulsar.broker.stats.prometheus.metrics;
import com.yahoo.sketches.quantiles.DoublesSketch;
-import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
import com.yahoo.sketches.quantiles.DoublesUnion;
import com.yahoo.sketches.quantiles.DoublesUnionBuilder;
-import io.netty.util.concurrent.FastThreadLocal;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
-import java.util.concurrent.locks.StampedLock;
public class DataSketchesSummaryLogger {
@@ -55,14 +50,7 @@ public class DataSketchesSummaryLogger {
countAdder.increment();
sumAdder.add((long) valueMillis);
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(valueMillis);
}
public void rotateLatencyCollection() {
@@ -72,15 +60,7 @@ public class DataSketchesSummaryLogger {
replacement = local;
final DoublesUnion aggregateValues = new DoublesUnionBuilder().build();
- local.map.forEach((localData, b) -> {
- long stamp = localData.lock.writeLock();
- try {
- aggregateValues.update(localData.successSketch);
- localData.successSketch.reset();
- } finally {
- localData.lock.unlockWrite(stamp);
- }
- });
+ local.record(aggregateValues, null);
values = aggregateValues.getResultAndReset();
}
@@ -97,27 +77,4 @@ public class DataSketchesSummaryLogger {
DoublesSketch s = values;
return s != null ? s.getQuantile(quantile) : Double.NaN;
}
-
- private static class LocalData {
- private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
- private final StampedLock lock = new StampedLock();
- }
-
- private static class ThreadLocalAccessor {
- private final Map<LocalData, Boolean> map = new ConcurrentHashMap<>();
- private final FastThreadLocal<LocalData> localData = new FastThreadLocal<LocalData>() {
-
- @Override
- protected LocalData initialValue() throws Exception {
- LocalData localData = new LocalData();
- map.put(localData, Boolean.TRUE);
- return localData;
- }
-
- @Override
- protected void onRemoval(LocalData value) throws Exception {
- map.remove(value);
- }
- };
- }
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
new file mode 100644
index 00000000000..3d5ca99bf8f
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
@@ -0,0 +1,134 @@
+/*
+ * 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.metrics;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.yahoo.sketches.quantiles.DoublesSketch;
+import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
+import com.yahoo.sketches.quantiles.DoublesUnion;
+import io.netty.util.concurrent.FastThreadLocal;
+import io.netty.util.concurrent.FastThreadLocalThread;
+import java.lang.ref.WeakReference;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.StampedLock;
+
+class ThreadLocalAccessor {
+
+ private final ConcurrentHashMap<LocalData, Boolean> map = new ConcurrentHashMap<>();
+ private final FastThreadLocal<LocalData> localData = new FastThreadLocal<>() {
+
+ @Override
+ protected LocalData initialValue() {
+ LocalData localData = new LocalData(Thread.currentThread());
+ map.put(localData, Boolean.TRUE);
+ return localData;
+ }
+
+ @Override
+ protected void onRemoval(LocalData value) {
+ map.remove(value);
+ }
+ };
+
+ void record(DoublesUnion aggregateSuccess, DoublesUnion aggregateFail) {
+ map.keySet().forEach(key -> {
+ key.record(aggregateSuccess, aggregateFail);
+ if (key.shouldRemove()) {
+ map.remove(key);
+ }
+ });
+ }
+
+ LocalData getLocalData() {
+ return localData.get();
+ }
+
+ @VisibleForTesting
+ int getLocalDataCount() {
+ return map.keySet().size();
+ }
+
+ static class LocalData {
+
+ private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
+ private final DoublesSketch failSketch = new DoublesSketchBuilder().build();
+ private final StampedLock lock = new StampedLock();
+ // Keep a weak reference to the owner thread so that we can remove the LocalData when the thread
+ // is not alive anymore or has been garbage collected.
+ // This reference isn't needed when the owner thread is a FastThreadLocalThread and will be null in that case.
+ // The removal is handled by FastThreadLocal#onRemoval when the owner thread is a FastThreadLocalThread.
+ private final WeakReference<Thread> ownerThreadReference;
+
+ LocalData(Thread ownerThread) {
+ if (ownerThread instanceof FastThreadLocalThread) {
+ ownerThreadReference = null;
+ } else {
+ ownerThreadReference = new WeakReference<>(ownerThread);
+ }
+ }
+
+ private boolean shouldRemove() {
+ if (ownerThreadReference == null) {
+ // the owner is a FastThreadLocalThread which handles the removal using FastThreadLocal#onRemoval
+ return false;
+ } else {
+ Thread ownerThread = ownerThreadReference.get();
+ if (ownerThread == null) {
+ // the thread has already been garbage collected, LocalData should be removed
+ return true;
+ } else {
+ // the thread isn't alive anymore, LocalData should be removed
+ return !ownerThread.isAlive();
+ }
+ }
+ }
+
+ void record(DoublesUnion aggregateSuccess, DoublesUnion aggregateFail) {
+ long stamp = lock.writeLock();
+ try {
+ aggregateSuccess.update(successSketch);
+ successSketch.reset();
+ if (aggregateFail != null) {
+ aggregateFail.update(failSketch);
+ failSketch.reset();
+ }
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ void updateSuccess(double value) {
+ long stamp = lock.readLock();
+ try {
+ successSketch.update(value);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ void updateFail(double value) {
+ long stamp = lock.readLock();
+ try {
+ failSketch.update(value);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
new file mode 100644
index 00000000000..7959b2181b8
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.metrics;
+
+import static org.testng.Assert.assertEquals;
+import com.yahoo.sketches.quantiles.DoublesUnion;
+import io.netty.util.concurrent.FastThreadLocalThread;
+import java.util.concurrent.Phaser;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ThreadLocalAccessorTest {
+
+ @DataProvider
+ public static Object[][] provider() {
+ return new Object[][] {
+ // 1st element: whether the thread is a FastThreadLocalThread
+ // 2nd element: the 2nd argument passed to the `ThreadLocalAccessor#record` method
+ { true, DoublesUnion.builder().build() },
+ { true, null },
+ { false, DoublesUnion.builder().build() },
+ { false, null },
+ };
+ }
+
+ @Test(dataProvider = "provider")
+ public void testShouldRemoveLocalDataWhenOwnerThreadIsNotAlive(
+ boolean fastThreadLocalThread, DoublesUnion aggregateFail) throws Exception {
+ // given a ThreadLocalAccessor instance
+ final var threadLocalAccessor = new ThreadLocalAccessor();
+ DoublesUnion aggregateSuccess = DoublesUnion.builder().build();
+ // using phaser to synchronize threads
+ Phaser phaser = new Phaser(2);
+ Thread thread = getThread(fastThreadLocalThread, () -> {
+ // Create a new LocalData instance for the current thread.
+ threadLocalAccessor.getLocalData();
+ // sync point #1, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // sync point #2, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ });
+ // sync point #1, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // and record is called
+ threadLocalAccessor.record(aggregateSuccess, aggregateFail);
+ // then LocalData should exist
+ assertEquals(threadLocalAccessor.getLocalDataCount(), 1);
+
+ // when thread is not alive anymore
+ // sync point #2, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // wait for thread to finish
+ thread.join();
+ // and record is called
+ threadLocalAccessor.record(aggregateSuccess, aggregateFail);
+ // then LocalData should be removed
+ assertEquals(threadLocalAccessor.getLocalDataCount(), 0);
+ }
+
+ @Test(dataProvider = "provider")
+ public void testThreadGc(boolean fastThreadLocalThread, DoublesUnion aggregateFail) throws Exception {
+ final var accessor = new ThreadLocalAccessor();
+ getThread(fastThreadLocalThread, accessor::getLocalData).join();
+ System.gc();
+ // FastThreadLocalThread removes the LocalData from the map when the thread finishes
+ assertEquals(accessor.getLocalDataCount(), fastThreadLocalThread ? 0 : 1);
+ accessor.record(DoublesUnion.builder().build(), aggregateFail);
+ assertEquals(accessor.getLocalDataCount(), 0);
+ }
+
+ private static Thread getThread(boolean fastThreadLocalThread, Runnable runnable) {
+ final var thread = fastThreadLocalThread ? new FastThreadLocalThread(runnable) : new Thread(runnable);
+ thread.start(); // when LocalData is created
+ return thread;
+ }
+}
(pulsar) 01/04: [improve][build] Disable javadoc build failure (#24594)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 26a31ac8a5c93ee369def2832ee6005c07a25e95
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Sep 8 22:22:17 2025 +0800
[improve][build] Disable javadoc build failure (#24594)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 783172fbe142e1a7e0d363a9a4e02ebff44f4141)
---
pom.xml | 2 ++
1 file changed, 2 insertions(+)
diff --git a/pom.xml b/pom.xml
index 34bfc83b3b1..1339b7ba61d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2119,6 +2119,7 @@ flexible messaging model and an intuitive client API.</description>
<configuration>
<doclint>none</doclint>
<notimestamp>true</notimestamp>
+ <failOnError>false</failOnError>
</configuration>
</plugin>
<plugin>
@@ -2366,6 +2367,7 @@ flexible messaging model and an intuitive client API.</description>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<doclint>none</doclint>
+ <failOnError>false</failOnError>
</configuration>
</plugin>
<plugin>