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 2026/10/17 06:51:24 UTC
(pulsar) branch branch-4.0 updated (1857ec204bc4 -> 0ca8e63bb1c6)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 1857ec204bc4 [fix][broker] Don't hang end txn requests when the error response has no message
new 5c300b776eee [fix][misc] Fix Backoff mandatory-stop timer tracking (#25282)
new 0ca8e63bb1c6 [fix][broker]The partitions of the topic was wrongly created even though this cluster was not allowed to access it (#25443)
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:
.../broker/admin/impl/PersistentTopicsBase.java | 46 +++++++----
.../pulsar/broker/service/BrokerService.java | 32 ++++++++
.../service/OneWayReplicatorUsingGlobalZKTest.java | 33 ++++++++
.../org/apache/pulsar/common/util/Backoff.java | 7 +-
.../org/apache/pulsar/common/util/BackoffTest.java | 93 ++++++++++++++++++++++
5 files changed, 195 insertions(+), 16 deletions(-)
(pulsar) 01/02: [fix][misc] Fix Backoff mandatory-stop timer tracking (#25282)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5c300b776eee1271344538cee37a55cb89a896af
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Oct 17 08:31:25 2026 +0800
[fix][misc] Fix Backoff mandatory-stop timer tracking (#25282)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 7c7efeed3cc6a19976cdd7616a868e00e56cd760)
---
.../org/apache/pulsar/common/util/Backoff.java | 7 +-
.../org/apache/pulsar/common/util/BackoffTest.java | 93 ++++++++++++++++++++++
2 files changed, 99 insertions(+), 1 deletion(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
index 842d3bbaa975..e4942d628c1a 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
@@ -35,6 +35,7 @@ public class Backoff {
private long mandatoryStop;
private long firstBackoffTimeInMillis;
+ private boolean firstBackoffTimeSet;
private boolean mandatoryStopMade = false;
private static final Random random = new Random();
@@ -50,6 +51,7 @@ public class Backoff {
this.mandatoryStop = unitMandatoryStop.toMillis(mandatoryStop);
this.clock = clock;
this.firstBackoffTimeInMillis = 0;
+ this.firstBackoffTimeSet = false;
}
public Backoff(long initial, TimeUnit unitInitial, long max, TimeUnit unitMax, long mandatoryStop,
@@ -67,8 +69,9 @@ public class Backoff {
if (!mandatoryStopMade) {
long now = clock.millis();
long timeElapsedSinceFirstBackoff = 0;
- if (initial == current) {
+ if (!firstBackoffTimeSet) {
firstBackoffTimeInMillis = now;
+ firstBackoffTimeSet = true;
} else {
timeElapsedSinceFirstBackoff = now - firstBackoffTimeInMillis;
}
@@ -95,6 +98,8 @@ public class Backoff {
public void reset() {
this.next = this.initial;
+ this.firstBackoffTimeInMillis = 0;
+ this.firstBackoffTimeSet = false;
if (initial == 0 && max == 0 && mandatoryStop == 0) {
this.mandatoryStopMade = true;
} else {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
index 9ca9503b5c09..2359fbd405fd 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
@@ -80,6 +80,99 @@ public class BackoffTest {
assertEquals(diffBackOffTime, 300);
}
+ @Test
+ public void mandatoryStopShouldWorkWithConstantBackoff() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(400L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 100, TimeUnit.MILLISECONDS,
+ 300, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // first call starts the mandatory-stop timer
+ backoff.next();
+ long firstBackoffTime = backoff.getFirstBackoffTimeInMillis();
+
+ // elapsed wall-clock time should exceed mandatory stop
+ backoff.next();
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), firstBackoffTime,
+ "firstBackoffTime should not be updated after the first call to next()");
+ assertTrue(backoff.isMandatoryStopMade(),
+ "mandatory stop should be reached even when initial == max (constant backoff)");
+ }
+
+ @Test
+ public void reduceToHalfShouldNotResetMandatoryStopTimer() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(200L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 400, TimeUnit.MILLISECONDS,
+ 250, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // first call starts the mandatory-stop timer (next becomes 200)
+ backoff.next();
+ long firstBackoffTime = backoff.getFirstBackoffTimeInMillis();
+
+ // This can bring the next delay back to initial, but it should not reset firstBackoffTime.
+ backoff.reduceToHalf();
+
+ backoff.next();
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), firstBackoffTime,
+ "reduceToHalf should not reset firstBackoffTime");
+ assertTrue(backoff.isMandatoryStopMade(),
+ "mandatory stop should be reached based on wall-clock time since the first call to next()");
+ }
+
+ @Test
+ public void resetShouldStartNewMandatoryStopCycle() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(400L)
+ .thenReturn(1000L)
+ .thenReturn(1400L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 100, TimeUnit.MILLISECONDS,
+ 300, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // Cycle 1: reach mandatory stop
+ backoff.next();
+ long firstCycleStart = backoff.getFirstBackoffTimeInMillis();
+ backoff.next();
+ assertTrue(backoff.isMandatoryStopMade(), "mandatory stop should be reached in cycle-1");
+
+ // Reset should clear mandatory-stop state so a new cycle can start.
+ backoff.reset();
+ assertFalse(backoff.isMandatoryStopMade(), "reset should clear mandatoryStopMade");
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0L, "reset should clear firstBackoffTime");
+
+ // Cycle 2: should start timing again and be able to reach mandatory stop again.
+ backoff.next();
+ long secondCycleStart = backoff.getFirstBackoffTimeInMillis();
+ assertEquals(secondCycleStart, 1000L,
+ "reset should start a new mandatory-stop timing window");
+ assertEquals(secondCycleStart - firstCycleStart, 1000L,
+ "reset should not reuse the old timing window");
+
+ backoff.next();
+ assertTrue(backoff.isMandatoryStopMade(), "mandatory stop should be reached in cycle-2");
+ }
+
@Test
public void basicTest() {
Clock mockClock = Clock.fixed(Instant.EPOCH, ZoneId.systemDefault());
(pulsar) 02/02: [fix][broker]The partitions of the topic was wrongly created even though this cluster was not allowed to access it (#25443)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0ca8e63bb1c6aeefdb6b69f0dd3a09dc53f65db2
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sat Oct 17 10:17:35 2026 +0800
[fix][broker]The partitions of the topic was wrongly created even though this cluster was not allowed to access it (#25443)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 1cd3457c0a52753a5e995b1b4a17e024cf41952f)
---
.../broker/admin/impl/PersistentTopicsBase.java | 46 +++++++++++++++-------
.../pulsar/broker/service/BrokerService.java | 32 +++++++++++++++
.../service/OneWayReplicatorUsingGlobalZKTest.java | 33 ++++++++++++++++
3 files changed, 96 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index dfea20089d79..49905421031a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -49,6 +49,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.ws.rs.WebApplicationException;
@@ -549,27 +550,42 @@ public class PersistentTopicsBase extends AdminResource {
}
protected void internalCreateMissedPartitions(AsyncResponse asyncResponse) {
- getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
- if (metadata != null && metadata.partitions > 0) {
- CompletableFuture<Void> future = validateNamespaceOperationAsync(topicName.getNamespaceObject(),
- NamespaceOperation.CREATE_TOPIC);
- future.thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> {
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(e -> {
- log.error("[{}] Failed to create partitions for topic {}", clientAppId(), topicName);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return null;
- });
- } else {
- throw new RestException(Status.NOT_FOUND, String.format("Topic %s does not exist", topicName));
- }
- }).exceptionally(ex -> {
+ Consumer<Throwable> errorHandler = ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to create partitions for topic {}",
clientAppId(), topicName);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
+ };
+ pulsar().getBrokerService().isCurrentClusterAllowed(topicName).thenAccept(allowed -> {
+ if (!allowed) {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.BAD_REQUEST, String.format("Topic [%s] is not allowed to be loaded"
+ + " up on the current cluster, please recheck replication polices.", topicName)));
+ return;
+ }
+ getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
+ if (metadata != null && metadata.partitions > 0) {
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC)
+ .thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> {
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ log.error("[{}] Failed to create partitions for topic {}", clientAppId(), topicName);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return null;
+ });
+ } else {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.NOT_FOUND, String.format("Topic %s does not exist", topicName)));
+ }
+ }).exceptionally(ex -> {
+ errorHandler.accept(ex);
+ return null;
+ });
+ }).exceptionally(ex -> {
+ errorHandler.accept(ex);
return null;
});
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index f487997f13e7..ab3e9c21a08b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -54,6 +54,7 @@ import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -4181,6 +4182,37 @@ public class BrokerService implements Closeable {
this.pulsarChannelInitFactory = factory;
}
+ /**
+ * Whether the current cluster is allowed to access the topic, by the same rule that is applied when the topic is
+ * loaded: the effective replication clusters (topic level, then global topic level, then namespace level) or the
+ * namespace-level allowed clusters contain the current cluster.
+ */
+ public CompletableFuture<Boolean> isCurrentClusterAllowed(@NonNull TopicName topicName) {
+ final String cluster = pulsar.getConfig().getClusterName();
+ final CompletableFuture<Optional<TopicPolicies>> topicPoliciesFuture =
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY);
+ final CompletableFuture<Optional<TopicPolicies>> globalTopicPoliciesFuture =
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.GLOBAL_ONLY);
+ final CompletableFuture<Optional<Policies>> nsPoliciesFuture = pulsar.getPulsarResources()
+ .getNamespaceResources().getPoliciesAsync(topicName.getNamespaceObject());
+ return CompletableFuture.allOf(topicPoliciesFuture, globalTopicPoliciesFuture, nsPoliciesFuture)
+ .thenApply(__ -> {
+ Optional<Policies> nsPolicies = nsPoliciesFuture.join();
+ List<String> topicClusters = topicPoliciesFuture.join()
+ .map(TopicPolicies::getReplicationClusters)
+ .or(() -> globalTopicPoliciesFuture.join().map(TopicPolicies::getReplicationClusters))
+ .orElse(null);
+ Collection<String> replicationClusters = topicClusters != null ? topicClusters
+ : nsPolicies.map(p -> p.replication_clusters).orElse(null);
+ // A topic without replication clusters is loaded without checking the cluster either.
+ if (isEmpty(replicationClusters)) {
+ return true;
+ }
+ return replicationClusters.contains(cluster)
+ || nsPolicies.map(p -> p.allowed_clusters.contains(cluster)).orElse(false);
+ });
+ }
+
/***
* After PIP-321 Introduce allowed-cluster at the namespace level, the condition that whether the cluster is
* allowed to access by the current cluster was defined by two fields:
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index 6e9879ee3454..b70adc28f6ce 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -190,6 +190,15 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
});
waitReplicatorStopped(subTopic, pulsar1, pulsar2, true);
+ try {
+ admin2.topics().createMissedPartitions(topicName);
+ fail("The action that creates mission partitions should have thrown exception");
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("is not allowed to be loaded up"));
+ }
+ assertFalse(admin2.topics().getList(replicatedNamespace)
+ .contains(TopicName.get(topicName).getPartition(0).toString()));
+
// Remove global policy.
admin1.topicPolicies(true).removeReplicationClusters(topicName);
Producer<byte[]> producer2 = client1.newProducer().topic(topicName).create();
@@ -952,4 +961,28 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
p.close();
admin1.topics().delete(topic);
}
+
+ /**
+ * A cluster in the namespace-level allowed clusters can load the topic even when the topic-level replication
+ * clusters do not contain it, so it is allowed to create the missed partitions too.
+ */
+ @Test
+ public void testCreateMissedPartitionsOnAllowedClusterOutsideTopicReplication() throws Exception {
+ final String ns1 = defaultTenant + "/ns_" + UUID.randomUUID().toString().replace("-", "");
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + ns1 + "/tp");
+ final String subTopic = TopicName.get(topicName).getPartition(0).toString();
+ admin1.namespaces().createNamespace(ns1);
+ admin1.namespaces().setNamespaceAllowedClusters(ns1, new HashSet<>(Arrays.asList(cluster1, cluster2)));
+ admin1.namespaces().setNamespaceReplicationClusters(ns1,
+ new HashSet<>(Arrays.asList(cluster1, cluster2)));
+ admin1.topics().createPartitionedTopic(topicName, 1);
+ admin1.topicPolicies(true).setReplicationClusters(topicName, Arrays.asList(cluster1));
+ Awaitility.await().untilAsserted(() -> {
+ Set<String> global2 = admin2.topicPolicies(true).getReplicationClusters(topicName, false);
+ assertEquals(global2, Collections.singleton(cluster1));
+ });
+
+ admin2.topics().createMissedPartitions(topicName);
+ assertTrue(admin2.topics().getList(ns1).contains(subTopic));
+ }
}