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 2024/03/27 23:06:09 UTC
(pulsar) branch branch-3.2 updated (f56c383a531 -> d9068698da4)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from f56c383a531 [fix][broker] Avoid expired unclosed ledgers when checking expired messages by ledger closure time (#22335)
new 622c87d1e82 [fix][broker] Check cursor state before adding it to the `waitingCursors` (#22191)
new d9068698da4 [improve][misc] Remove the call to sun InetAddressCachePolicy (#22329)
The 2 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:
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 10 +++++
.../service/persistent/PersistentTopicTest.java | 46 +++++++++++++++++++
.../pulsar/common/util/netty/DnsResolverUtil.java | 52 ++++++++++++++++------
.../pulsar/common/util/netty/DnsResolverTest.java | 44 ++++++++++++++++++
5 files changed, 139 insertions(+), 15 deletions(-)
(pulsar) 01/02: [fix][broker] Check cursor state before adding it to the `waitingCursors` (#22191)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 622c87d1e82163c8b42281c0c7a40e3ad79dc0df
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Thu Mar 28 06:53:21 2024 +0800
[fix][broker] Check cursor state before adding it to the `waitingCursors` (#22191)
(cherry picked from commit b702d440dc5e5a4cfd845bf60d5e310efe665ff5)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 10 +++++
.../service/persistent/PersistentTopicTest.java | 46 ++++++++++++++++++++++
3 files changed, 57 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index da013c07313..0da71e89018 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -987,7 +987,7 @@ public class ManagedCursorImpl implements ManagedCursor {
name);
}
// Let the managed ledger know we want to be notified whenever a new entry is published
- ledger.waitingCursors.add(this);
+ ledger.addWaitingCursor(this);
} else {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Skip notification registering since we do have entries available",
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 75ac4dd4c0a..ee8e7c430ef 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -3803,6 +3803,16 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
this.waitingCursors.remove(cursor);
}
+ public void addWaitingCursor(ManagedCursorImpl cursor) {
+ if (cursor instanceof NonDurableCursorImpl) {
+ if (cursor.isActive()) {
+ this.waitingCursors.add(cursor);
+ }
+ } else {
+ this.waitingCursors.add(cursor);
+ }
+ }
+
public boolean isCursorActive(ManagedCursor cursor) {
return activeCursors.get(cursor.getName()) != null;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
index 717dfc28ac8..618fe8006a6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
@@ -57,10 +57,15 @@ import java.util.function.Supplier;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorContainer;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.TopicPoliciesService;
import org.apache.pulsar.broker.stats.PrometheusMetricsTest;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
@@ -74,6 +79,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.TopicName;
@@ -661,4 +667,44 @@ public class PersistentTopicTest extends BrokerTestBase {
subscribe.close();
admin.topics().delete(topicName);
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors";
+ admin.topics().createNonPartitionedTopic(topicName);
+ final Optional<Topic> topic = pulsar.getBrokerService().getTopic(topicName, false).join();
+ assertNotNull(topic.get());
+ PersistentTopic persistentTopic = (PersistentTopic) topic.get();
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl)persistentTopic.getManagedLedger();
+ final ManagedCursor spyCursor= spy(ledger.newNonDurableCursor(PositionImpl.LATEST, "sub-2"));
+ doAnswer((invocation) -> {
+ Thread.sleep(10_000);
+ invocation.callRealMethod();
+ return null;
+ }).when(spyCursor).asyncReadEntriesOrWait(any(int.class), any(long.class),
+ any(AsyncCallbacks.ReadEntriesCallback.class), any(Object.class), any(PositionImpl.class));
+ Field cursorField = ManagedLedgerImpl.class.getDeclaredField("cursors");
+ cursorField.setAccessible(true);
+ ManagedCursorContainer container = (ManagedCursorContainer) cursorField.get(ledger);
+ container.removeCursor("sub-2");
+ container.add(spyCursor, null);
+ final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.NonDurable)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("sub-2").subscribe();
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ producer.send("test");
+ producer.close();
+ final Message<String> receive = consumer.receive();
+ assertEquals("test", receive.getValue());
+ consumer.close();
+ Awaitility.await()
+ .pollDelay(5, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ assertEquals(ledger.getWaitingCursorsCount(), 0);
+ });
+ }
}
(pulsar) 02/02: [improve][misc] Remove the call to sun InetAddressCachePolicy (#22329)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d9068698da48f826cea2c8d5075ce95497cecc94
Author: Jian Yun <yi...@gmail.com>
AuthorDate: Thu Mar 28 06:59:28 2024 +0800
[improve][misc] Remove the call to sun InetAddressCachePolicy (#22329)
Co-authored-by: Lari Hotari <lh...@apache.org>
(cherry picked from commit cce0b058efd55e2d5ac42c4ecaceddacee648a7c)
---
.../pulsar/common/util/netty/DnsResolverUtil.java | 52 ++++++++++++++++------
.../pulsar/common/util/netty/DnsResolverTest.java | 44 ++++++++++++++++++
2 files changed, 82 insertions(+), 14 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
index f49a6453c72..bcff83acd94 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
@@ -19,12 +19,20 @@
package org.apache.pulsar.common.util.netty;
import io.netty.resolver.dns.DnsNameResolverBuilder;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
+import java.security.Security;
+import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class DnsResolverUtil {
+
+ private static final String CACHE_POLICY_PROP = "networkaddress.cache.ttl";
+ private static final String CACHE_POLICY_PROP_FALLBACK = "sun.net.inetaddr.ttl";
+ private static final String NEGATIVE_CACHE_POLICY_PROP = "networkaddress.cache.negative.ttl";
+ private static final String NEGATIVE_CACHE_POLICY_PROP_FALLBACK = "sun.net.inetaddr.negative.ttl";
+ /* default ttl value from sun.net.InetAddressCachePolicy.DEFAULT_POSITIVEļ¼ which is used when no security manager
+ is used */
+ private static final int JDK_DEFAULT_TTL = 30;
private static final int MIN_TTL = 0;
private static final int TTL;
private static final int NEGATIVE_TTL;
@@ -39,19 +47,35 @@ public class DnsResolverUtil {
int ttl = DEFAULT_TTL;
int negativeTtl = DEFAULT_NEGATIVE_TTL;
try {
- // use reflection to call sun.net.InetAddressCachePolicy's get and getNegative methods for getting
- // effective JDK settings for DNS caching
- Class<?> inetAddressCachePolicyClass = Class.forName("sun.net.InetAddressCachePolicy");
- Method getTTLMethod = inetAddressCachePolicyClass.getMethod("get");
- ttl = (Integer) getTTLMethod.invoke(null);
- Method getNegativeTTLMethod = inetAddressCachePolicyClass.getMethod("getNegative");
- negativeTtl = (Integer) getNegativeTTLMethod.invoke(null);
- } catch (NoSuchMethodException | ClassNotFoundException | InvocationTargetException
- | IllegalAccessException e) {
- log.warn("Cannot get DNS TTL settings from sun.net.InetAddressCachePolicy class", e);
+ String ttlStr = Security.getProperty(CACHE_POLICY_PROP);
+ if (ttlStr == null) {
+ // Compatible with sun.net.inetaddr.ttl settings
+ ttlStr = System.getProperty(CACHE_POLICY_PROP_FALLBACK);
+ }
+ String negativeTtlStr = Security.getProperty(NEGATIVE_CACHE_POLICY_PROP);
+ if (negativeTtlStr == null) {
+ // Compatible with sun.net.inetaddr.negative.ttl settings
+ negativeTtlStr = System.getProperty(NEGATIVE_CACHE_POLICY_PROP_FALLBACK);
+ }
+ ttl = Optional.ofNullable(ttlStr)
+ .map(Integer::decode)
+ .filter(i -> i > 0)
+ .orElseGet(() -> {
+ if (System.getSecurityManager() == null) {
+ return JDK_DEFAULT_TTL;
+ }
+ return DEFAULT_TTL;
+ });
+
+ negativeTtl = Optional.ofNullable(negativeTtlStr)
+ .map(Integer::decode)
+ .filter(i -> i >= 0)
+ .orElse(DEFAULT_NEGATIVE_TTL);
+ } catch (NumberFormatException e) {
+ log.warn("Cannot get DNS TTL settings", e);
}
- TTL = ttl <= 0 ? DEFAULT_TTL : ttl;
- NEGATIVE_TTL = negativeTtl < 0 ? DEFAULT_NEGATIVE_TTL : negativeTtl;
+ TTL = ttl;
+ NEGATIVE_TTL = negativeTtl;
}
private DnsResolverUtil() {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
index 0ccb960e798..46599cc45a0 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
@@ -18,13 +18,57 @@
*/
package org.apache.pulsar.common.util.netty;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
import io.netty.channel.EventLoop;
import io.netty.resolver.dns.DnsNameResolverBuilder;
+import java.security.Security;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
public class DnsResolverTest {
+ private static final int MIN_TTL = 0;
+ private static final int TTL = 101;
+ private static final int NEGATIVE_TTL = 121;
+ private static final String CACHE_POLICY_PROP = "networkaddress.cache.ttl";
+ private static final String NEGATIVE_CACHE_POLICY_PROP = "networkaddress.cache.negative.ttl";
+
+ private String originalCachePolicy;
+ private String originalNegativeCachePolicy;
+
+ @BeforeClass(alwaysRun = true)
+ public void beforeClass() {
+ originalCachePolicy = Security.getProperty(CACHE_POLICY_PROP);
+ originalNegativeCachePolicy = Security.getProperty(NEGATIVE_CACHE_POLICY_PROP);
+ Security.setProperty(CACHE_POLICY_PROP, Integer.toString(TTL));
+ Security.setProperty(NEGATIVE_CACHE_POLICY_PROP, Integer.toString(NEGATIVE_TTL));
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void afterClass() {
+ Security.setProperty(CACHE_POLICY_PROP, originalCachePolicy != null ? originalCachePolicy : "-1");
+ Security.setProperty(NEGATIVE_CACHE_POLICY_PROP,
+ originalNegativeCachePolicy != null ? originalNegativeCachePolicy : "0");
+ }
+
+ @Test
+ public void testTTl() {
+ final DnsNameResolverBuilder builder = mock(DnsNameResolverBuilder.class);
+ ArgumentCaptor<Integer> minTtlCaptor = ArgumentCaptor.forClass(Integer.class);
+ ArgumentCaptor<Integer> maxTtlCaptor = ArgumentCaptor.forClass(Integer.class);
+ ArgumentCaptor<Integer> negativeTtlCaptor = ArgumentCaptor.forClass(Integer.class);
+ DnsResolverUtil.applyJdkDnsCacheSettings(builder);
+ verify(builder).ttl(minTtlCaptor.capture(), maxTtlCaptor.capture());
+ verify(builder).negativeTtl(negativeTtlCaptor.capture());
+ assertEquals(minTtlCaptor.getValue(), MIN_TTL);
+ assertEquals(maxTtlCaptor.getValue(), TTL);
+ assertEquals(negativeTtlCaptor.getValue(), NEGATIVE_TTL);
+ }
@Test
public void testMaxTtl() {