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:24:08 UTC
(pulsar) branch branch-3.3 updated (c88d33fb843 -> 6bd945830f4)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from c88d33fb843 [fix][sec] Upgrade Netty to 4.1.127.Final to address CVEs (#24717)
new 66295c62df5 [improve][build] Disable javadoc build failure (#24594)
new cd38a3205b7 [fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
new 2665ad011bf [feat][fn] Fallback to using `STATE_STORAGE_SERVICE_URL` in `PulsarMetadataStateStoreProviderImpl.init` (#24721)
new 98c641e3022 [fix][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads (#24725)
new 6bd945830f4 [fix][io] Improve Kafka Connect source offset flushing logic (#24654)
The 5 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 | 135 +++++++++++++
.../metrics/ThreadLocalAccessorTest.java | 93 +++++++++
.../PersistentAcknowledgmentsGroupingTracker.java | 29 +--
.../impl/AcknowledgementsGroupingTrackerTest.java | 28 +--
.../PulsarMetadataStateStoreProviderImpl.java | 4 +
pulsar-io/kafka-connect-adaptor/pom.xml | 6 +
.../kafka/connect/AbstractKafkaConnectSource.java | 141 +++++++------
.../io/kafka/connect/KafkaConnectSourceTest.java | 186 ++++++++++++++++-
.../functions/PulsarBKStateStoreTest.java} | 8 +
.../functions/PulsarMetadataStateStoreTest.java | 60 ++++++
.../integration/functions/PulsarStateTest.java | 220 ++++++++++++---------
.../src/test/resources/pulsar-function.xml | 3 +-
15 files changed, 741 insertions(+), 310 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
copy tests/{docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/functions/package-info.java => integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarBKStateStoreTest.java} (77%)
create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarMetadataStateStoreTest.java
(pulsar) 03/05: [feat][fn] Fallback to using `STATE_STORAGE_SERVICE_URL` in `PulsarMetadataStateStoreProviderImpl.init` (#24721)
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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2665ad011bfaa717b4fda10f3516110595d05634
Author: jiangpengcheng <sc...@gmail.com>
AuthorDate: Wed Sep 10 19:37:36 2025 +0800
[feat][fn] Fallback to using `STATE_STORAGE_SERVICE_URL` in `PulsarMetadataStateStoreProviderImpl.init` (#24721)
(cherry picked from commit 415c6fa74e908bad2da1e7f986185c6ef17cb9a1)
---
.../PulsarMetadataStateStoreProviderImpl.java | 4 +
.../functions/PulsarBKStateStoreTest.java | 27 +++
.../functions/PulsarMetadataStateStoreTest.java | 60 ++++++
.../integration/functions/PulsarStateTest.java | 220 ++++++++++++---------
.../src/test/resources/pulsar-function.xml | 3 +-
5 files changed, 225 insertions(+), 89 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/state/PulsarMetadataStateStoreProviderImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/state/PulsarMetadataStateStoreProviderImpl.java
index 7b9807b2a78..5bffc5966b6 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/state/PulsarMetadataStateStoreProviderImpl.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/state/PulsarMetadataStateStoreProviderImpl.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.functions.instance.state;
import java.util.Map;
import lombok.SneakyThrows;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
@@ -46,6 +47,9 @@ public class PulsarMetadataStateStoreProviderImpl implements StateStoreProvider
shouldCloseStore = false;
} else {
String metadataUrl = (String) config.get(METADATA_URL);
+ if (StringUtils.isEmpty(metadataUrl)) {
+ metadataUrl = (String) config.get(StateStoreProvider.STATE_STORAGE_SERVICE_URL);
+ }
store = MetadataStoreFactory.create(metadataUrl, MetadataStoreConfig.builder()
.metadataStoreName(MetadataStoreConfig.STATE_METADATA_STORE).build());
shouldCloseStore = true;
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarBKStateStoreTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarBKStateStoreTest.java
new file mode 100644
index 00000000000..482dddb46d7
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarBKStateStoreTest.java
@@ -0,0 +1,27 @@
+/*
+ * 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.tests.integration.functions;
+
+import org.apache.pulsar.functions.instance.state.BKStateStoreProviderImpl;
+
+public class PulsarBKStateStoreTest extends PulsarStateTest {
+ protected PulsarBKStateStoreTest() {
+ super(BKStateStoreProviderImpl.class.getName());
+ }
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarMetadataStateStoreTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarMetadataStateStoreTest.java
new file mode 100644
index 00000000000..cb745928914
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarMetadataStateStoreTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.tests.integration.functions;
+
+import static org.testng.Assert.assertEquals;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.instance.state.PulsarMetadataStateStoreProviderImpl;
+import org.apache.pulsar.tests.integration.containers.PulsarContainer;
+import org.apache.pulsar.tests.integration.containers.StandaloneContainer;
+import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase;
+import org.testcontainers.containers.Network;
+
+@Slf4j
+public class PulsarMetadataStateStoreTest extends PulsarStateTest {
+ protected PulsarMetadataStateStoreTest() {
+ super(PulsarMetadataStateStoreProviderImpl.class.getName());
+ }
+
+ public void setUpCluster() throws Exception {
+ incrementSetupNumber();
+ network = Network.newNetwork();
+ String clusterName = PulsarClusterTestBase.randomName(8);
+ container = new StandaloneContainer(clusterName, PulsarContainer.DEFAULT_IMAGE_NAME)
+ .withNetwork(network)
+ .withNetworkAliases(StandaloneContainer.NAME + "-" + clusterName)
+ .withEnv("PULSAR_STANDALONE_USE_ZOOKEEPER", "true")
+ .withEnv("PF_stateStorageProviderImplementation", PulsarMetadataStateStoreProviderImpl.class.getName())
+ .withEnv("PF_stateStorageServiceUrl", "zk:localhost:2181");
+ container.start();
+ log.info("Pulsar cluster {} is up running:", clusterName);
+ log.info("\tBinary Service Url : {}", container.getPlainTextServiceUrl());
+ log.info("\tHttp Service Url : {}", container.getHttpServiceUrl());
+
+ // add cluster to public tenant
+ ContainerExecResult result = container.execCmd(
+ "/pulsar/bin/pulsar-admin", "namespaces", "policies", "public/default");
+ assertEquals(0, result.getExitCode());
+ log.info("public/default namespace policies are {}", result.getStdout());
+ }
+
+
+}
+
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
index 856e4edfea0..2395c32ed39 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
@@ -44,6 +44,8 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.functions.FunctionState;
import org.apache.pulsar.common.policies.data.SinkStatus;
import org.apache.pulsar.common.policies.data.SourceStatus;
+import org.apache.pulsar.functions.api.examples.WordCountFunction;
+import org.apache.pulsar.functions.instance.state.PulsarMetadataStateStoreProviderImpl;
import org.apache.pulsar.tests.integration.docker.ContainerExecException;
import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator;
@@ -57,37 +59,49 @@ import org.testng.annotations.Test;
* State related test cases.
*/
@Slf4j
-public class PulsarStateTest extends PulsarStandaloneTestSuite {
+public abstract class PulsarStateTest extends PulsarStandaloneTestSuite {
+
+ protected String stateStoreProvider;
+
+ protected PulsarStateTest(String stateStoreProvider) {
+ super();
+ this.stateStoreProvider = stateStoreProvider;
+ }
public static final String WORDCOUNT_PYTHON_CLASS =
- "wordcount_function.WordCountFunction";
+ "wordcount_function.WordCountFunction";
+ public static final String WORDCOUNT_JAVA_CLASS = WordCountFunction.class.getName();
public static final String WORDCOUNT_PYTHON_FILE = "wordcount_function.py";
public static final String VALUE_BASE64 = "0a8001127e0a172e6576656e74732e437573746f6d65724372656174656412630a243"
- + "2336366666263652d623038342d346631352d616565342d326330643135356131666"
- + "36312026e311a3700000000000000000000000000000000000000000000000000000"
- + "000000000000000000000000000000000000000000000000000000000";
-
- @Test(groups = {"python_state", "state", "function", "python_function"})
- public void testPythonWordCountFunction() throws Exception {
- String functionName = "test-wordcount-py-fn-" + randomName(8);
- doTestPythonWordCountFunction(functionName);
-
- // after a function is deleted, its state should be clean
- // we just recreate and test the word count function again, and it should have same result
- doTestPythonWordCountFunction(functionName);
- }
+ + "2336366666263652d623038342d346631352d616565342d326330643135356131666"
+ + "36312026e311a3700000000000000000000000000000000000000000000000000000"
+ + "000000000000000000000000000000000000000000000000000000000";
private void doTestPythonWordCountFunction(String functionName) throws Exception {
String inputTopicName = "test-wordcount-py-input-" + randomName(8);
String outputTopicName = "test-wordcount-py-output-" + randomName(8);
- final int numMessages = 10;
- // submit the exclamation function
- submitExclamationFunction(
+ // submit the word count function
+ submitWordCountFunction(
Runtime.PYTHON, inputTopicName, outputTopicName, functionName);
+ // run tests
+ doTestFunction(functionName, inputTopicName, outputTopicName);
+ }
+
+ private void doTestJavaWordCountFunction(String functionName) throws Exception {
+ String inputTopicName = "test-wordcount-java-input-" + randomName(8);
+
+ // submit the word count function
+ submitWordCountFunction(
+ Runtime.JAVA, inputTopicName, null, functionName);
+ // run tests
+ doTestFunction(functionName, inputTopicName, null);
+ }
+ private void doTestFunction(String functionName, String inputTopicName, String outputTopicName) throws Exception {
+ final int numMessages = 10;
// get function info
getFunctionInfoSuccess(functionName);
@@ -134,6 +148,31 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
getFunctionInfoNotFound(functionName);
}
+ @Test(groups = {"python_state", "state", "function", "python_function"})
+ public void testPythonWordCountFunction() throws Exception {
+ if (PulsarMetadataStateStoreProviderImpl.class.getName().equals(stateStoreProvider)) {
+ // python function doesn't support metadata state store yet
+ return;
+ }
+ String functionName = "test-wordcount-py-fn-" + randomName(8);
+ doTestPythonWordCountFunction(functionName);
+
+ // after a function is deleted, its state should be clean
+ // we just recreate and test the word count function again, and it should have same result
+ doTestPythonWordCountFunction(functionName);
+ }
+
+ @Test(groups = {"java_state", "state", "function", "java_function"})
+ public void testJavaWordCountFunction() throws Exception {
+ String functionName = "test-wordcount-java-fn-" + randomName(8);
+ doTestJavaWordCountFunction(functionName);
+
+ // after a function is deleted, its state should be clean
+ // we just recreate and test the word count function again, and it should have same result
+ doTestJavaWordCountFunction(functionName);
+ }
+
+
@Test(groups = {"java_state", "state", "function", "java_function"})
public void testSourceState() throws Exception {
String outputTopicName = "test-state-source-output-" + randomName(8);
@@ -195,7 +234,7 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
String sinkName = "test-state-sink-" + randomName(8);
int numMessages = 10;
- submitSinkConnector(sinkName, inputTopicName, "org.apache.pulsar.tests.integration.io.TestStateSink", JAVAJAR);
+ submitSinkConnector(sinkName, inputTopicName, "org.apache.pulsar.tests.integration.io.TestStateSink", JAVAJAR);
// get sink info
getSinkInfoSuccess(sinkName);
@@ -320,9 +359,9 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
}
private void submitSourceConnector(String sourceName,
- String outputTopicName,
- String className,
- String archive) throws Exception {
+ String outputTopicName,
+ String className,
+ String archive) throws Exception {
String[] commands = {
PulsarCluster.ADMIN_SCRIPT,
"sources", "create",
@@ -339,9 +378,9 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
}
private void submitSinkConnector(String sinkName,
- String inputTopicName,
- String className,
- String archive) throws Exception {
+ String inputTopicName,
+ String className,
+ String archive) throws Exception {
String[] commands = {
PulsarCluster.ADMIN_SCRIPT,
"sinks", "create",
@@ -357,33 +396,35 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
result.getStdout());
}
- private void submitExclamationFunction(Runtime runtime,
- String inputTopicName,
- String outputTopicName,
- String functionName) throws Exception {
+ private void submitWordCountFunction(Runtime runtime,
+ String inputTopicName,
+ String outputTopicName,
+ String functionName) throws Exception {
submitFunction(
- runtime,
- inputTopicName,
- outputTopicName,
- functionName,
- getExclamationClass(runtime),
- Schema.BYTES);
+ runtime,
+ inputTopicName,
+ outputTopicName,
+ functionName,
+ getWordCountClass(runtime),
+ Schema.BYTES);
}
- protected static String getExclamationClass(Runtime runtime) {
+ protected static String getWordCountClass(Runtime runtime) {
if (Runtime.PYTHON == runtime) {
return WORDCOUNT_PYTHON_CLASS;
+ } else if (Runtime.JAVA == runtime) {
+ return WORDCOUNT_JAVA_CLASS;
} else {
throw new IllegalArgumentException("Unsupported runtime : " + runtime);
}
}
private <T> void submitFunction(Runtime runtime,
- String inputTopicName,
- String outputTopicName,
- String functionName,
- String functionClass,
- Schema<T> inputTopicSchema) throws Exception {
+ String inputTopicName,
+ String outputTopicName,
+ String functionName,
+ String functionClass,
+ Schema<T> inputTopicSchema) throws Exception {
CommandGenerator generator;
generator = CommandGenerator.createDefaultGenerator(inputTopicName, functionClass);
generator.setSinkTopic(outputTopicName);
@@ -398,10 +439,10 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
throw new IllegalArgumentException("Unsupported runtime : " + runtime);
}
String[] commands = {
- "sh", "-c", command
+ "sh", "-c", command
};
ContainerExecResult result = container.execCmd(
- commands);
+ commands);
assertTrue(result.getStdout().contains("Created successfully"));
ensureSubscriptionCreated(inputTopicName, String.format("public/default/%s", functionName), inputTopicSchema);
@@ -409,18 +450,18 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
@SuppressWarnings("try")
private <T> void ensureSubscriptionCreated(String inputTopicName,
- String subscriptionName,
- Schema<T> inputTopicSchema)
+ String subscriptionName,
+ Schema<T> inputTopicSchema)
throws Exception {
// ensure the function subscription exists before we start producing messages
try (PulsarClient client = PulsarClient.builder()
- .serviceUrl(container.getPlainTextServiceUrl())
- .build()) {
+ .serviceUrl(container.getPlainTextServiceUrl())
+ .build()) {
try (Consumer<T> ignored = client.newConsumer(inputTopicSchema)
- .topic(inputTopicName)
- .subscriptionType(SubscriptionType.Shared)
- .subscriptionName(subscriptionName)
- .subscribe()) {
+ .topic(inputTopicName)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName(subscriptionName)
+ .subscribe()) {
}
}
}
@@ -451,12 +492,12 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
private void getFunctionInfoSuccess(String functionName) throws Exception {
ContainerExecResult result = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "functions",
- "get",
- "--tenant", "public",
- "--namespace", "default",
- "--name", functionName
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "get",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName
);
assertTrue(result.getStdout().contains("\"name\": \"" + functionName + "\""));
}
@@ -502,27 +543,27 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
private void getFunctionStatus(String functionName, int numMessages) throws Exception {
ContainerExecResult result = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "functions",
- "getstatus",
- "--tenant", "public",
- "--namespace", "default",
- "--name", functionName
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "getstatus",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName
);
assertTrue(result.getStdout().contains("\"running\" : true"));
assertTrue(result.getStdout().contains("\"numSuccessfullyProcessed\" : " + numMessages));
}
private void queryState(String functionName, String key, int amount, long version)
- throws Exception {
+ throws Exception {
ContainerExecResult result = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "functions",
- "querystate",
- "--tenant", "public",
- "--namespace", "default",
- "--name", functionName,
- "--key", key
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "querystate",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName,
+ "--key", key
);
assertTrue(result.getStdout().contains("\"numberValue\": " + amount));
assertTrue(result.getStdout().contains("\"version\": " + version));
@@ -582,24 +623,27 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
}
private void publishAndConsumeMessages(String inputTopic,
- String outputTopic,
- int numMessages) throws Exception {
+ String outputTopic,
+ int numMessages) throws Exception {
@Cleanup PulsarClient client = PulsarClient.builder()
- .serviceUrl(container.getPlainTextServiceUrl())
- .build();
- @Cleanup Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
- .topic(outputTopic)
- .subscriptionType(SubscriptionType.Exclusive)
- .subscriptionName("test-sub")
- .subscribe();
+ .serviceUrl(container.getPlainTextServiceUrl())
+ .build();
@Cleanup Producer<byte[]> producer = client.newProducer(Schema.BYTES)
- .topic(inputTopic)
- .create();
+ .topic(inputTopic)
+ .create();
for (int i = 0; i < numMessages; i++) {
producer.send(("hello test message-" + i).getBytes(UTF_8));
}
+ if (outputTopic == null) {
+ return;
+ }
+ @Cleanup Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
+ .topic(outputTopic)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("test-sub")
+ .subscribe();
for (int i = 0; i < numMessages; i++) {
Message<byte[]> msg = consumer.receive();
assertEquals("hello test message-" + i + "!", new String(msg.getValue(), UTF_8));
@@ -608,12 +652,12 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
private void deleteFunction(String functionName) throws Exception {
ContainerExecResult result = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "functions",
- "delete",
- "--tenant", "public",
- "--namespace", "default",
- "--name", functionName
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "delete",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName
);
assertTrue(result.getStdout().contains("Deleted successfully"));
result.assertNoStderr();
diff --git a/tests/integration/src/test/resources/pulsar-function.xml b/tests/integration/src/test/resources/pulsar-function.xml
index a18a65ff2e1..61c7b5b4698 100644
--- a/tests/integration/src/test/resources/pulsar-function.xml
+++ b/tests/integration/src/test/resources/pulsar-function.xml
@@ -22,7 +22,8 @@
<suite name="Pulsar Function Integration Tests" verbose="2" annotations="JDK">
<test name="pulsar-function-test-suite" preserve-order="true" >
<classes>
- <class name="org.apache.pulsar.tests.integration.functions.PulsarStateTest" />
+ <class name="org.apache.pulsar.tests.integration.functions.PulsarMetadataStateStoreTest" />
+ <class name="org.apache.pulsar.tests.integration.functions.PulsarBKStateStoreTest" />
<class name="org.apache.pulsar.tests.integration.io.PulsarGenericObjectSinkTest"/>
<class name="org.apache.pulsar.tests.integration.io.sources.GenericRecordSourceTest" />
<class name="org.apache.pulsar.tests.integration.io.sources.PulsarSourcePropertyTest"/>
(pulsar) 05/05: [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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6bd945830f4f627d01231b511bb2dbd2301e42f3
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 16541481dd8..c8139fac03c 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) 02/05: [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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cd38a3205b7278efda4bd5ebd634aa60bcbff691
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 | 135 +++++++++++++++++++++
.../metrics/ThreadLocalAccessorTest.java | 93 ++++++++++++++
4 files changed, 239 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..6a32ce5b905
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
@@ -0,0 +1,135 @@
+/*
+ * 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;
+import org.jspecify.annotations.Nullable;
+
+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, @Nullable 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, @Nullable 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..94c8337307d
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.jspecify.annotations.Nullable;
+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, @Nullable 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, @Nullable 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) 04/05: [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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 98c641e3022481657275e2bb0e2111855feae05f
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 | 28 +++++++--------------
2 files changed, 24 insertions(+), 33 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 5f7957d7f1d..b814d261fd7 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 7be113dcf5e..986229518fd 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,24 +31,20 @@ 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.impl.metrics.InstrumentProvider;
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;
@@ -389,24 +385,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 ++) {
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) 01/05: [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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 66295c62df51717828770e03b166f2a448b72949
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 36215ed0e84..10c83ecb914 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2224,6 +2224,7 @@ flexible messaging model and an intuitive client API.</description>
<configuration>
<doclint>none</doclint>
<notimestamp>true</notimestamp>
+ <failOnError>false</failOnError>
</configuration>
</plugin>
<plugin>
@@ -2471,6 +2472,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>