You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2023/07/05 05:43:24 UTC
[pulsar] branch branch-3.0 updated: Revert "[fix][broker] Fix NPE when reset Replicator's cursor by position. (#20597)"
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 35b449b6baf Revert "[fix][broker] Fix NPE when reset Replicator's cursor by position. (#20597)"
35b449b6baf is described below
commit 35b449b6baf036da6caa8beb1d54e58aca6c654f
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Jul 5 13:42:59 2023 +0800
Revert "[fix][broker] Fix NPE when reset Replicator's cursor by position. (#20597)"
This reverts commit aa565c21c8103be8edbafaffa90029f90bce37da.
---
.../broker/admin/impl/PersistentTopicsBase.java | 42 ++++++++--------------
.../persistent/PersistentMessageExpiryMonitor.java | 16 ++++-----
.../service/persistent/PersistentReplicator.java | 2 +-
.../service/persistent/PersistentSubscription.java | 2 +-
.../service/PersistentMessageFinderTest.java | 20 +++--------
.../pulsar/broker/service/ReplicatorTest.java | 30 ----------------
6 files changed, 28 insertions(+), 84 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 03e1a1b2747..81f5e3c1f32 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
@@ -4145,43 +4145,31 @@ public class PersistentTopicsBase extends AdminResource {
return;
}
try {
- PersistentSubscription sub = null;
- PersistentReplicator repl = null;
-
- if (subName.startsWith(topic.getReplicatorPrefix())) {
- String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
- repl = (PersistentReplicator)
- topic.getPersistentReplicator(remoteCluster);
- if (repl == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- "Replicator not found"));
- return;
- }
- } else {
- sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
- return;
- }
+ PersistentSubscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND,
+ getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ return;
}
-
CompletableFuture<Integer> batchSizeFuture = new CompletableFuture<>();
getEntryBatchSize(batchSizeFuture, topic, messageId, batchIndex);
-
- PersistentReplicator finalRepl = repl;
- PersistentSubscription finalSub = sub;
-
batchSizeFuture.thenAccept(bi -> {
PositionImpl position = calculatePositionAckSet(isExcluded, bi, batchIndex, messageId);
boolean issued;
try {
if (subName.startsWith(topic.getReplicatorPrefix())) {
- issued = finalRepl.expireMessages(position);
+ String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
+ PersistentReplicator repl = (PersistentReplicator)
+ topic.getPersistentReplicator(remoteCluster);
+ if (repl == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND,
+ "Replicator not found"));
+ return;
+ }
+ issued = repl.expireMessages(position);
} else {
- issued = finalSub.expireMessages(position);
+ issued = sub.expireMessages(position);
}
-
if (issued) {
log.info("[{}] Message expire started up to {} on {} {}", clientAppId(), position,
topicName, subName);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
index 3acd683f9f4..bddcb1b334d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -22,7 +22,6 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
-import javax.annotation.Nullable;
import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -44,7 +43,6 @@ import org.slf4j.LoggerFactory;
public class PersistentMessageExpiryMonitor implements FindEntryCallback {
private final ManagedCursor cursor;
private final String subName;
- private final PersistentTopic topic;
private final String topicName;
private final Rate msgExpired;
private final LongAdder totalMsgExpired;
@@ -59,10 +57,9 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback {
expirationCheckInProgressUpdater = AtomicIntegerFieldUpdater
.newUpdater(PersistentMessageExpiryMonitor.class, "expirationCheckInProgress");
- public PersistentMessageExpiryMonitor(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
- @Nullable PersistentSubscription subscription) {
- this.topic = topic;
- this.topicName = topic.getName();
+ public PersistentMessageExpiryMonitor(String topicName, String subscriptionName, ManagedCursor cursor,
+ PersistentSubscription subscription) {
+ this.topicName = topicName;
this.cursor = cursor;
this.subName = subscriptionName;
this.subscription = subscription;
@@ -101,12 +98,11 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback {
public boolean expireMessages(Position messagePosition) {
// If it's beyond last position of this topic, do nothing.
- PositionImpl topicLastPosition = (PositionImpl) this.topic.getLastPosition();
- if (topicLastPosition.compareTo((PositionImpl) messagePosition) < 0) {
+ if (((PositionImpl) subscription.getTopic().getLastPosition()).compareTo((PositionImpl) messagePosition) < 0) {
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Ignore expire-message scheduled task, given position {} is beyond "
- + "current topic's last position {}", topicName, subName, messagePosition,
- topicLastPosition);
+ + "current topic's last position {}", topicName, subName, messagePosition,
+ subscription.getTopic().getLastPosition());
}
return false;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index d468cb0643f..ccf70eecec3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -116,7 +116,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
brokerService, replicationClient);
this.topic = localTopic;
this.cursor = cursor;
- this.expiryMonitor = new PersistentMessageExpiryMonitor(localTopic,
+ this.expiryMonitor = new PersistentMessageExpiryMonitor(localTopicName,
Codec.decode(cursor.getName()), cursor, null);
HAVE_PENDING_READ_UPDATER.set(this, FALSE);
PENDING_MESSAGES_UPDATER.set(this, 0);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 192e5ed3e00..090e8e25377 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -152,7 +152,7 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
this.topicName = topic.getName();
this.subName = subscriptionName;
this.fullName = MoreObjects.toStringHelper(this).add("topic", topicName).add("name", subName).toString();
- this.expiryMonitor = new PersistentMessageExpiryMonitor(topic, subscriptionName, cursor, this);
+ this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
this.setReplicated(replicated);
this.subscriptionProperties = MapUtils.isEmpty(subscriptionProperties)
? Collections.emptyMap() : Collections.unmodifiableMap(subscriptionProperties);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index f0e2e6eafcd..e77fd07c6ef 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -62,7 +62,6 @@ import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.broker.service.persistent.PersistentMessageExpiryMonitor;
import org.apache.pulsar.broker.service.persistent.PersistentMessageFinder;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
-import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.impl.ResetCursorData;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
@@ -231,11 +230,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
});
assertTrue(ex.get());
- PersistentTopic mock = mock(PersistentTopic.class);
- when(mock.getName()).thenReturn("topicname");
- when(mock.getLastPosition()).thenReturn(PositionImpl.EARLIEST);
-
- PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(mock, c1.getName(), c1, null);
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor("topicname", c1.getName(), c1, null);
monitor.findEntryFailed(new ManagedLedgerException.ConcurrentFindCursorPositionException("failed"),
Optional.empty(), null);
Field field = monitor.getClass().getDeclaredField("expirationCheckInProgress");
@@ -412,11 +407,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
bkc.deleteLedger(ledgers.get(1).getLedgerId());
bkc.deleteLedger(ledgers.get(2).getLedgerId());
- PersistentTopic mock = mock(PersistentTopic.class);
- when(mock.getName()).thenReturn("topicname");
- when(mock.getLastPosition()).thenReturn(PositionImpl.EARLIEST);
-
- PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(mock, c1.getName(), c1, null);
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor("topicname", c1.getName(), c1, null);
Position previousMarkDelete = null;
for (int i = 0; i < totalEntries; i++) {
monitor.expireMessages(1);
@@ -453,16 +444,15 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
PersistentSubscription subscription = mock(PersistentSubscription.class);
- PersistentTopic topic = mock(PersistentTopic.class);
+ Topic topic = mock(Topic.class);
when(subscription.getTopic()).thenReturn(topic);
- when(topic.getName()).thenReturn("topicname");
for (int i = 0; i < totalEntries; i++) {
positions.add(ledger.addEntry(createMessageWrittenToLedger("msg" + i)));
}
when(topic.getLastPosition()).thenReturn(positions.get(positions.size() - 1));
- PersistentMessageExpiryMonitor monitor = spy(new PersistentMessageExpiryMonitor(topic,
+ PersistentMessageExpiryMonitor monitor = spy(new PersistentMessageExpiryMonitor("topicname",
cursor.getName(), cursor, subscription));
assertEquals(cursor.getMarkDeletedPosition(), PositionImpl.get(positions.get(0).getLedgerId(), -1));
boolean issued;
@@ -501,7 +491,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
clearInvocations(monitor);
ManagedCursorImpl mockCursor = mock(ManagedCursorImpl.class);
- PersistentMessageExpiryMonitor mockMonitor = spy(new PersistentMessageExpiryMonitor(topic,
+ PersistentMessageExpiryMonitor mockMonitor = spy(new PersistentMessageExpiryMonitor("topicname",
cursor.getName(), mockCursor, subscription));
// Not calling findEntryComplete to clear expirationCheckInProgress condition, so following call to
// expire message shouldn't issue.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index f710c8541d1..176eab0e94b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -728,36 +728,6 @@ public class ReplicatorTest extends ReplicatorTestBase {
assertEquals(status.getReplicationBacklog(), 0);
}
-
- @Test(timeOut = 30000)
- public void testResetReplicatorSubscriptionPosition() throws Exception {
- final TopicName dest = TopicName
- .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/resetReplicatorSubscription"));
-
- @Cleanup
- MessageProducer producer1 = new MessageProducer(url1, dest);
-
- // Produce from cluster1 and consume from the rest
- for (int i = 0; i < 10; i++) {
- producer1.produce(2);
- }
-
- PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get();
-
- PersistentReplicator replicator = (PersistentReplicator) spy(
- topic.getReplicators().get(topic.getReplicators().keys().get(0)));
-
- MessageId id = topic.getLastMessageId().get();
- admin1.topics().expireMessages(dest.getPartitionedTopicName(),
- replicator.getCursor().getName(),
- id,false);
-
- replicator.updateRates();
-
- ReplicatorStats status = replicator.getStats();
- assertEquals(status.getReplicationBacklog(), 0);
- }
-
@Test(timeOut = 30000)
public void testResetCursorNotFail() throws Exception {