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 2025/07/07 14:15:41 UTC
(pulsar) branch branch-3.0 updated (8d007ddb6f4 -> 1d493967b8f)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 8d007ddb6f4 [fix][txn] Fix negative unacknowledged messages in transactions by ensuring that the batch size is added into CommandAck (#24443)
new e0cc63a2222 [fix][broker] Fix ack hole in cursor for geo-replication (#20931)
new 76417d0e8c3 [Fix][broker] Limit replication rate based on bytes (#22674)
new 1d493967b8f [fix][broker]excessive replication speed leads to error: Producer send queue is full (#24189)
The 3 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:
.../pulsar/broker/service/AbstractReplicator.java | 31 +-
.../apache/pulsar/broker/service/Replicator.java | 2 +-
.../nonpersistent/NonPersistentReplicator.java | 5 +
.../persistent/GeoPersistentReplicator.java | 34 +-
.../service/persistent/MessageDeduplication.java | 5 +-
.../service/persistent/PersistentReplicator.java | 430 ++++++++++++++++-----
.../broker/service/persistent/PersistentTopic.java | 11 +-
.../service/persistent/ShadowReplicator.java | 16 +-
.../broker/service/AbstractReplicatorTest.java | 3 +
.../BrokerServiceInternalMethodInvoker.java} | 7 +-
.../broker/service/OneWayReplicatorTest.java | 153 +++++++-
.../broker/service/OneWayReplicatorTestBase.java | 26 +-
...OneWayReplicatorUsingGlobalPartitionedTest.java | 12 +
.../service/OneWayReplicatorUsingGlobalZKTest.java | 14 +-
.../broker/service/ReplicatorRateLimiterTest.java | 60 +++
.../broker/service/ReplicatorSubscriptionTest.java | 16 +-
.../pulsar/broker/service/ReplicatorTest.java | 29 +-
.../BrokerServicePersistInternalMethodInvoker.java | 47 +++
.../PersistentReplicatorInflightTaskTest.java | 367 ++++++++++++++++++
.../service/persistent/ShadowReplicatorTest.java | 19 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 13 +-
.../client/impl/GeoReplicationProducerImpl.java | 8 +
.../apache/pulsar/client/impl/ProducerImpl.java | 17 +
23 files changed, 1174 insertions(+), 151 deletions(-)
copy pulsar-broker/src/{main/java/org/apache/pulsar/broker/service/TopicPolicyListener.java => test/java/org/apache/pulsar/broker/service/BrokerServiceInternalMethodInvoker.java} (81%)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BrokerServicePersistInternalMethodInvoker.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
(pulsar) 02/03: [Fix][broker] Limit replication rate based on bytes (#22674)
Posted by yu...@apache.org.
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
commit 76417d0e8c329732db6b72d26def868e9a02862d
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Jun 28 23:52:08 2024 +0800
[Fix][broker] Limit replication rate based on bytes (#22674)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 7f4c0c535971d5b85c48a9cd658ae0e28dc46932)
---
.../service/persistent/PersistentReplicator.java | 79 ++++++++++++++--------
.../broker/service/ReplicatorRateLimiterTest.java | 60 ++++++++++++++++
2 files changed, 111 insertions(+), 28 deletions(-)
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 81afff2012c..1ea24d54142 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
@@ -33,6 +33,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import lombok.AllArgsConstructor;
+import lombok.Data;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
@@ -214,15 +216,31 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
}
+ @Data
+ @AllArgsConstructor
+ private static class AvailablePermits {
+ private int messages;
+ private long bytes;
+
+ /**
+ * messages, bytes
+ * 0, O: Producer queue is full, no permits.
+ * -1, -1: Rate Limiter reaches limit.
+ * >0, >0: available permits for read entries.
+ */
+ public boolean isExceeded() {
+ return messages == -1 && bytes == -1;
+ }
+
+ public boolean isReadable() {
+ return messages > 0 && bytes > 0;
+ }
+ }
+
/**
* Calculate available permits for read entries.
- *
- * @return
- * 0: Producer queue is full, no permits.
- * -1: Rate Limiter reaches limit.
- * >0: available permits for read entries.
*/
- private int getAvailablePermits() {
+ private AvailablePermits getAvailablePermits() {
int availablePermits = producerQueueSize - PENDING_MESSAGES_UPDATER.get(this);
// return 0, if Producer queue is full, it will pause read entries.
@@ -231,14 +249,20 @@ public abstract class PersistentReplicator extends AbstractReplicator
log.debug("[{}] Producer queue is full, availablePermits: {}, pause reading",
replicatorId, availablePermits);
}
- return 0;
+ return new AvailablePermits(0, 0);
}
+ long availablePermitsOnMsg = -1;
+ long availablePermitsOnByte = -1;
+
// handle rate limit
if (dispatchRateLimiter.isPresent() && dispatchRateLimiter.get().isDispatchRateLimitingEnabled()) {
DispatchRateLimiter rateLimiter = dispatchRateLimiter.get();
+ // if dispatch-rate is in msg then read only msg according to available permit
+ availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
+ availablePermitsOnByte = rateLimiter.getAvailableDispatchRateLimitOnByte();
// no permits from rate limit
- if (!rateLimiter.hasMessageDispatchPermit()) {
+ if (availablePermitsOnByte == 0 || availablePermitsOnMsg == 0) {
if (log.isDebugEnabled()) {
log.debug("[{}] message-read exceeded topic replicator message-rate {}/{},"
+ " schedule after a {}",
@@ -247,17 +271,18 @@ public abstract class PersistentReplicator extends AbstractReplicator
rateLimiter.getDispatchRateOnByte(),
MESSAGE_RATE_BACKOFF_MS);
}
- return -1;
- }
-
- // if dispatch-rate is in msg then read only msg according to available permit
- long availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
- if (availablePermitsOnMsg > 0) {
- availablePermits = Math.min(availablePermits, (int) availablePermitsOnMsg);
+ return new AvailablePermits(-1, -1);
}
}
- return availablePermits;
+ availablePermitsOnMsg =
+ availablePermitsOnMsg == -1 ? availablePermits : Math.min(availablePermits, availablePermitsOnMsg);
+ availablePermitsOnMsg = Math.min(availablePermitsOnMsg, readBatchSize);
+
+ availablePermitsOnByte =
+ availablePermitsOnByte == -1 ? readMaxSizeBytes : Math.min(readMaxSizeBytes, availablePermitsOnByte);
+
+ return new AvailablePermits((int) availablePermitsOnMsg, availablePermitsOnByte);
}
protected void readMoreEntries() {
@@ -265,10 +290,10 @@ public abstract class PersistentReplicator extends AbstractReplicator
log.info("[{}] Skip the reading due to new detected schema", replicatorId);
return;
}
- int availablePermits = getAvailablePermits();
-
- if (availablePermits > 0) {
- int messagesToRead = Math.min(availablePermits, readBatchSize);
+ AvailablePermits availablePermits = getAvailablePermits();
+ if (availablePermits.isReadable()) {
+ int messagesToRead = availablePermits.getMessages();
+ long bytesToRead = availablePermits.getBytes();
if (!isWritable()) {
if (log.isDebugEnabled()) {
log.debug("[{}] Throttling replication traffic because producer is not writable", replicatorId);
@@ -277,9 +302,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
messagesToRead = 1;
}
- // If messagesToRead is 0 or less, correct it to 1 to prevent IllegalArgumentException
- messagesToRead = Math.max(messagesToRead, 1);
-
// Schedule read
if (HAVE_PENDING_READ_UPDATER.compareAndSet(this, FALSE, TRUE)) {
if (waitForCursorRewinding) {
@@ -288,17 +310,18 @@ public abstract class PersistentReplicator extends AbstractReplicator
return;
}
if (log.isDebugEnabled()) {
- log.debug("[{}] Schedule read of {} messages", replicatorId, messagesToRead);
+ log.debug("[{}] Schedule read of {} messages or {} bytes", replicatorId, messagesToRead,
+ bytesToRead);
}
- cursor.asyncReadEntriesOrWait(messagesToRead, readMaxSizeBytes, this,
+ cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this,
null, topic.getMaxReadPosition());
} else {
if (log.isDebugEnabled()) {
- log.debug("[{}] Not scheduling read due to pending read. Messages To Read {}",
- replicatorId, messagesToRead);
+ log.debug("[{}] Not scheduling read due to pending read. Messages To Read {}, Bytes To Read {}",
+ replicatorId, messagesToRead, bytesToRead);
}
}
- } else if (availablePermits == -1) {
+ } else if (availablePermits.isExceeded()) {
// no permits from rate limit
topic.getBrokerService().executor().schedule(
() -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
index b97210b009a..d92bc6005fd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -545,5 +546,64 @@ public class ReplicatorRateLimiterTest extends ReplicatorTestBase {
producer.close();
}
+ @Test
+ public void testReplicatorRateLimiterByBytes() throws Exception {
+ final String namespace = "pulsar/replicatormsg-" + System.currentTimeMillis();
+ final String topicName = "persistent://" + namespace + "/RateLimiterByBytes";
+
+ admin1.namespaces().createNamespace(namespace);
+ // 0. set 2 clusters, there will be 1 replicator in each topic
+ admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2"));
+
+ final int byteRate = 400;
+ final int payloadSize = 100;
+ DispatchRate dispatchRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(-1)
+ .dispatchThrottlingRateInByte(byteRate)
+ .ratePeriodInSecond(360)
+ .build();
+ admin1.namespaces().setReplicatorDispatchRate(namespace, dispatchRate);
+
+ @Cleanup
+ PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).build();
+ @Cleanup
+ Producer<byte[]> producer = client1.newProducer().topic(topicName)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+
+ PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getOrCreateTopic(topicName).get();
+
+ Awaitility.await()
+ .untilAsserted(() -> assertTrue(topic.getReplicators().values().get(0).getRateLimiter().isPresent()));
+ assertEquals(topic.getReplicators().values().get(0).getRateLimiter().get().getDispatchRateOnByte(), byteRate);
+
+ @Cleanup
+ PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString())
+ .build();
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+
+ @Cleanup
+ Consumer<byte[]> ignored = client2.newConsumer().topic(topicName).subscriptionName("sub2-in-cluster2")
+ .messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ }).subscribe();
+
+ // The total bytes is 5 times the rate limit value.
+ int numMessages = byteRate / payloadSize * 5;
+ for (int i = 0; i < numMessages * payloadSize; i++) {
+ producer.send(new byte[payloadSize]);
+ }
+
+ Awaitility.await().pollDelay(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ // The rate limit occurs in the next reading cycle, so a value fault tolerance needs to be added.
+ assertThat(totalReceived.get()).isLessThan((byteRate / payloadSize) + 2);
+ });
+ }
+
private static final Logger log = LoggerFactory.getLogger(ReplicatorRateLimiterTest.class);
}
(pulsar) 03/03: [fix][broker]excessive replication speed leads to error: Producer send queue is full (#24189)
Posted by yu...@apache.org.
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
commit 1d493967b8f0997c7bdc944aafdec15e76b1c749
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Jul 2 14:11:25 2025 +0800
[fix][broker]excessive replication speed leads to error: Producer send queue is full (#24189)
Co-authored-by: Kai Wang <kw...@streamnative.io>
(cherry picked from commit 37b17d3e6920e851fa56f962b43bd698963033f5)
---
.../pulsar/broker/service/AbstractReplicator.java | 31 +-
.../apache/pulsar/broker/service/Replicator.java | 2 +-
.../nonpersistent/NonPersistentReplicator.java | 5 +
.../persistent/GeoPersistentReplicator.java | 34 +-
.../service/persistent/MessageDeduplication.java | 5 +-
.../service/persistent/PersistentReplicator.java | 376 ++++++++++++++++-----
.../broker/service/persistent/PersistentTopic.java | 11 +-
.../service/persistent/ShadowReplicator.java | 16 +-
.../broker/service/AbstractReplicatorTest.java | 3 +
.../BrokerServiceInternalMethodInvoker.java} | 36 +-
.../broker/service/OneWayReplicatorTest.java | 153 ++++++++-
.../broker/service/OneWayReplicatorTestBase.java | 26 +-
...OneWayReplicatorUsingGlobalPartitionedTest.java | 12 +
.../service/OneWayReplicatorUsingGlobalZKTest.java | 14 +-
.../broker/service/ReplicatorSubscriptionTest.java | 16 +-
.../pulsar/broker/service/ReplicatorTest.java | 29 +-
.../BrokerServicePersistInternalMethodInvoker.java | 47 +++
.../PersistentReplicatorInflightTaskTest.java | 367 ++++++++++++++++++++
.../service/persistent/ShadowReplicatorTest.java | 19 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 13 +-
.../client/impl/GeoReplicationProducerImpl.java | 8 +
.../apache/pulsar/client/impl/ProducerImpl.java | 17 +
22 files changed, 1064 insertions(+), 176 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
index 9677ed52e3c..d52e0336e60 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -286,9 +286,10 @@ public abstract class AbstractReplicator implements Replicator {
/**
* This method only be used by {@link PersistentTopic#checkGC} now.
*/
- public CompletableFuture<Void> disconnect(boolean failIfHasBacklog, boolean closeTheStartingProducer) {
+ @Override
+ public CompletableFuture<Void> disconnect() {
long backlog = getNumberOfEntriesInBacklog();
- if (failIfHasBacklog && backlog > 0) {
+ if (backlog > 0) {
CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
disconnectFuture.completeExceptionally(new TopicBusyException("Cannot close a replicator with backlog"));
if (log.isDebugEnabled()) {
@@ -298,9 +299,30 @@ public abstract class AbstractReplicator implements Replicator {
}
log.info("[{}] Disconnect replicator at position {} with backlog {}", replicatorId,
getReplicatorReadPosition(), backlog);
- return closeProducerAsync(closeTheStartingProducer);
+ return beforeDisconnect()
+ .thenCompose(__ -> closeProducerAsync(true))
+ .thenApply(__ -> {
+ afterDisconnected();
+ return null;
+ });
}
+ /**
+ * This method and {@link #afterDisconnected()} are used to solve the following race condition:
+ * - Thread 1: calling disconnect.
+ * passed the check: no backlog.
+ * - Thread 2: published a message, then the cursor.pendingRead completes.
+ * - Thread 1: continue to disconnect.
+ * - Thread 2: read entries from the cursor, and try to send messages, but the messages will be discarded because
+ * the producer is closed.
+ * Issue: the pending reading's read position is not correct.
+ */
+ protected CompletableFuture<Void> beforeDisconnect() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ protected void afterDisconnected() {}
+
/**
* This method only be used by {@link PersistentTopic#checkGC} now.
*/
@@ -379,12 +401,15 @@ public abstract class AbstractReplicator implements Replicator {
});
}
+ protected abstract void beforeTerminate();
+
public CompletableFuture<Void> terminate() {
if (!tryChangeStatusToTerminating()) {
log.info("[{}] Skip current termination since other thread is doing termination, state : {}", replicatorId,
state);
return CompletableFuture.completedFuture(null);
}
+ beforeTerminate();
return doCloseProducerAsync(producer, () -> {
STATE_UPDATER.set(this, State.Terminated);
this.producer = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java
index 5c314397da8..2573a389c95 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java
@@ -31,7 +31,7 @@ public interface Replicator {
CompletableFuture<Void> terminate();
- CompletableFuture<Void> disconnect(boolean failIfHasBacklog, boolean closeTheStartingProducer);
+ CompletableFuture<Void> disconnect();
void updateRates();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
index 51509f3818a..b5bedd1bf42 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
@@ -262,4 +262,9 @@ public class NonPersistentReplicator extends AbstractReplicator implements Repli
ProducerImpl<?> producer = this.producer;
return producer != null && producer.isConnected();
}
+
+ @Override
+ protected void beforeTerminate() {
+ // No-op
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
index aed609a7743..934c81828aa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
@@ -82,7 +82,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
}
@Override
- protected boolean replicateEntries(List<Entry> entries) {
+ protected boolean replicateEntries(List<Entry> entries, final InFlightTask inFlightTask) {
boolean atLeastOneMessageSentForReplication = false;
boolean isEnableReplicatedSubscriptions =
brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions();
@@ -91,12 +91,13 @@ public class GeoPersistentReplicator extends PersistentReplicator {
// This flag is set to true when we skip at least one local message,
// in order to skip remaining local messages.
boolean isLocalMessageSkippedOnce = false;
- boolean skipRemainingMessages = false;
+ boolean skipRemainingMessages = inFlightTask.isSkipReadResultDueToCursorRewind();
for (int i = 0; i < entries.size(); i++) {
Entry entry = entries.get(i);
// Skip the messages since the replicator need to fetch the schema info to replicate the schema to the
// remote cluster. Rewind the cursor first and continue the message read after fetched the schema.
if (skipRemainingMessages) {
+ inFlightTask.incCompletedEntries();
entry.release();
continue;
}
@@ -109,12 +110,14 @@ public class GeoPersistentReplicator extends PersistentReplicator {
log.error("[{}] Failed to deserialize message at {} (buffer size: {}): {}", replicatorId,
entry.getPosition(), length, t.getMessage(), t);
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
continue;
}
if (Markers.isTxnMarker(msg.getMessageBuilder())) {
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -124,6 +127,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
msg.getMessageBuilder().getTxnidLeastBits());
if (topic.isTxnAborted(tx, (PositionImpl) entry.getPosition())) {
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -137,6 +141,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
if (msg.isReplicated()) {
// Discard messages that were already replicated into this region
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -148,6 +153,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
entry.getPosition(), msg.getReplicateTo());
}
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -160,6 +166,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
replicatorId, entry.getPosition(), msg.getReplicateTo());
}
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -167,12 +174,13 @@ public class GeoPersistentReplicator extends PersistentReplicator {
if (STATE_UPDATER.get(this) != State.Started || isLocalMessageSkippedOnce) {
// The producer is not ready yet after having stopped/restarted. Drop the message because it will
- // recovered when the producer is ready
+ // recover when the producer is ready
if (log.isDebugEnabled()) {
log.debug("[{}] Dropping read message at {} because producer is not ready",
replicatorId, entry.getPosition());
}
isLocalMessageSkippedOnce = true;
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -185,14 +193,23 @@ public class GeoPersistentReplicator extends PersistentReplicator {
CompletableFuture<SchemaInfo> schemaFuture = getSchemaInfo(msg);
if (!schemaFuture.isDone() || schemaFuture.isCompletedExceptionally()) {
+ /**
+ * Skip in flight reading tasks.
+ * Explain the result of the race-condition between:
+ * - {@link #readMoreEntries}
+ * - {@link #beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding)}
+ * Since {@link #acquirePermitsIfNotFetchingSchema} and
+ * {@link #beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding)} acquire the
+ * same lock, it is safe.
+ */
+ beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Fetching_Schema);
+ inFlightTask.incCompletedEntries();
entry.release();
headersAndPayload.release();
msg.recycle();
// Mark the replicator is fetching the schema for now and rewind the cursor
// and trigger the next read after complete the schema fetching.
- fetchSchemaInProgress = true;
skipRemainingMessages = true;
- cursor.cancelPendingReadRequest();
log.info("[{}] Pause the data replication due to new detected schema", replicatorId);
schemaFuture.whenComplete((__, e) -> {
if (e != null) {
@@ -200,9 +217,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
replicatorId, e);
}
log.info("[{}] Resume the data replication after the schema fetching done", replicatorId);
- cursor.rewind();
- fetchSchemaInProgress = false;
- readMoreEntries();
+ doRewindCursor(true);
});
} else {
msg.setSchemaInfoForReplicator(schemaFuture.get());
@@ -213,11 +228,10 @@ public class GeoPersistentReplicator extends PersistentReplicator {
.setValue(String.format("%s:%s", entry.getLedgerId(), entry.getEntryId()));
msgOut.recordEvent(headersAndPayload.readableBytes());
// Increment pending messages for messages produced locally
- PENDING_MESSAGES_UPDATER.incrementAndGet(this);
if (log.isDebugEnabled()) {
log.debug("[{}] Publishing {}:{}", replicatorId, entry.getLedgerId(), entry.getEntryId());
}
- producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg));
+ producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg, inFlightTask));
atLeastOneMessageSentForReplication = true;
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index 1463ee83158..1bb4bdf2bbd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -96,8 +96,9 @@ public class MessageDeduplication {
}
public static class MessageDupUnknownException extends RuntimeException {
- public MessageDupUnknownException() {
- super("Cannot determine whether the message is a duplicate at this time");
+ public MessageDupUnknownException(String topicName, String producerName) {
+ super(String.format("[%s][%s]Cannot determine whether the message is a duplicate at this time", topicName,
+ producerName));
}
}
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 1ea24d54142..b48d31ec364 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
@@ -27,12 +27,13 @@ import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
+import java.util.Collections;
+import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
@@ -48,10 +49,12 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.CursorAlreadyClosedE
import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.service.AbstractReplicator;
import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.Replicator;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.client.api.MessageId;
@@ -84,19 +87,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
private final int producerQueueThreshold;
- protected static final AtomicIntegerFieldUpdater<PersistentReplicator> PENDING_MESSAGES_UPDATER =
- AtomicIntegerFieldUpdater
- .newUpdater(PersistentReplicator.class, "pendingMessages");
- private volatile int pendingMessages = 0;
-
- private static final int FALSE = 0;
- private static final int TRUE = 1;
-
- private static final AtomicIntegerFieldUpdater<PersistentReplicator> HAVE_PENDING_READ_UPDATER =
- AtomicIntegerFieldUpdater
- .newUpdater(PersistentReplicator.class, "havePendingRead");
- private volatile int havePendingRead = FALSE;
-
protected final Rate msgOut = new Rate();
protected final Rate msgExpired = new Rate();
@@ -111,8 +101,17 @@ public abstract class PersistentReplicator extends AbstractReplicator
private final ReplicatorStatsImpl stats = new ReplicatorStatsImpl();
- protected volatile boolean fetchSchemaInProgress = false;
- private volatile boolean waitForCursorRewinding = false;
+ protected volatile int waitForCursorRewindingRefCnf = 0;
+
+ protected enum ReasonOfWaitForCursorRewinding {
+ Failed_Publishing,
+ Fetching_Schema,
+ Disconnecting,
+ Terminating;
+ }
+ protected ReasonOfWaitForCursorRewinding reasonOfWaitForCursorRewinding = null;
+
+ protected final LinkedList<InFlightTask> inFlightTasks = new LinkedList<>();
public PersistentReplicator(String localCluster, PersistentTopic localTopic, ManagedCursor cursor,
String remoteCluster, String remoteTopic,
@@ -124,8 +123,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
this.cursor = cursor;
this.expiryMonitor = new PersistentMessageExpiryMonitor(localTopicName,
Codec.decode(cursor.getName()), cursor, null);
- HAVE_PENDING_READ_UPDATER.set(this, FALSE);
- PENDING_MESSAGES_UPDATER.set(this, 0);
readBatchSize = Math.min(
producerQueueSize,
@@ -140,11 +137,8 @@ public abstract class PersistentReplicator extends AbstractReplicator
@Override
protected void setProducerAndTriggerReadEntries(Producer<byte[]> producer) {
- waitForCursorRewinding = true;
-
// Repeat until there are no read operations in progress
- if (STATE_UPDATER.get(this) == State.Starting && HAVE_PENDING_READ_UPDATER.get(this) == TRUE
- && !cursor.cancelPendingReadRequest()) {
+ if (STATE_UPDATER.get(this) == State.Starting && hasPendingRead() && !cursor.cancelPendingReadRequest()) {
brokerService.getPulsar().getExecutor()
.schedule(() -> setProducerAndTriggerReadEntries(producer), 10, TimeUnit.MILLISECONDS);
return;
@@ -161,12 +155,10 @@ public abstract class PersistentReplicator extends AbstractReplicator
if (!(producer instanceof ProducerImpl)) {
log.error("[{}] The partitions count between two clusters is not the same, the replicator can not be"
+ " created successfully: {}", replicatorId, state);
- waitForCursorRewinding = false;
doCloseProducerAsync(producer, () -> {});
throw new ClassCastException(producer.getClass().getName() + " can not be cast to ProducerImpl");
}
this.producer = (ProducerImpl) producer;
- HAVE_PENDING_READ_UPDATER.set(this, FALSE);
// Trigger a new read.
log.info("[{}] Created replicator producer, Replicator state: {}", replicatorId, state);
backOff.reset();
@@ -175,8 +167,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
// Rewind the cursor to be sure to read again all non-acked messages sent while restarting
cursor.rewind();
- waitForCursorRewinding = false;
-
// read entries
readMoreEntries();
} else {
@@ -192,7 +182,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
log.error("[{}] Replicator state is not expected, so close the producer. Replicator state: {}",
replicatorId, changeStateRes.getRight());
}
- waitForCursorRewinding = false;
// Close the producer if change the state fail.
doCloseProducerAsync(producer, () -> {});
}
@@ -240,8 +229,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
/**
* Calculate available permits for read entries.
*/
- private AvailablePermits getAvailablePermits() {
- int availablePermits = producerQueueSize - PENDING_MESSAGES_UPDATER.get(this);
+ private AvailablePermits getRateLimiterAvailablePermits(int availablePermits) {
// return 0, if Producer queue is full, it will pause read entries.
if (availablePermits <= 0) {
@@ -286,51 +274,59 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
protected void readMoreEntries() {
- if (fetchSchemaInProgress) {
- log.info("[{}] Skip the reading due to new detected schema", replicatorId);
- return;
- }
- AvailablePermits availablePermits = getAvailablePermits();
- if (availablePermits.isReadable()) {
- int messagesToRead = availablePermits.getMessages();
- long bytesToRead = availablePermits.getBytes();
- if (!isWritable()) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Throttling replication traffic because producer is not writable", replicatorId);
- }
- // Minimize the read size if the producer is disconnected or the window is already full
- messagesToRead = 1;
+ // Acquire permits and check state of producer.
+ InFlightTask newInFlightTask = acquirePermitsIfNotFetchingSchema();
+ if (newInFlightTask == null) {
+ // no permits from rate limit
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Not scheduling read due to pending read or no permits.", replicatorId);
}
-
- // Schedule read
- if (HAVE_PENDING_READ_UPDATER.compareAndSet(this, FALSE, TRUE)) {
- if (waitForCursorRewinding) {
- log.info("[{}] Skip the reading because repl producer is starting", replicatorId);
- HAVE_PENDING_READ_UPDATER.set(this, FALSE);
- return;
- }
- if (log.isDebugEnabled()) {
- log.debug("[{}] Schedule read of {} messages or {} bytes", replicatorId, messagesToRead,
- bytesToRead);
- }
- cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this,
- null, topic.getMaxReadPosition());
+ if (!hasPendingRead()) {
+ topic.getBrokerService().executor().schedule(
+ () -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
+ return;
} else {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Not scheduling read due to pending read. Messages To Read {}, Bytes To Read {}",
- replicatorId, messagesToRead, bytesToRead);
- }
+ return;
+ }
+ }
+ // If disabled RateLimiter.
+ if (!dispatchRateLimiter.isPresent() || !dispatchRateLimiter.get().isDispatchRateLimitingEnabled()) {
+ cursor.asyncReadEntriesOrWait(newInFlightTask.readingEntries, -1, this,
+ newInFlightTask/* Context object */, topic.getMaxReadPosition());
+ return;
+ }
+ // No permits of RateLimiter.
+ AvailablePermits availablePermits = getRateLimiterAvailablePermits(newInFlightTask.readingEntries);
+ if (!availablePermits.isReadable()) {
+ // no rate limiter permits from rate limit
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Throttling replication traffic. Messages To Read {}, Bytes To Read {}",
+ replicatorId, availablePermits.getMessages(), availablePermits.getBytes());
}
- } else if (availablePermits.isExceeded()) {
- // no permits from rate limit
topic.getBrokerService().executor().schedule(
- () -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
- } else {
+ () -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
+ return;
+ }
+ // Has permits of RateLimiter.
+ int messagesToRead = availablePermits.getMessages();
+ long bytesToRead = availablePermits.getBytes();
+ if (!isWritable()) {
if (log.isDebugEnabled()) {
- log.debug("[{}] No Permits for reading. availablePermits: {}",
- replicatorId, availablePermits);
+ log.debug("[{}] Throttling replication traffic because producer is not writable", replicatorId);
}
+ // Minimize the read size if the producer is disconnected or the window is already full
+ messagesToRead = 1;
+ }
+ // Update acquired permits exceeds limitation.
+ if (messagesToRead < newInFlightTask.readingEntries) {
+ newInFlightTask.setReadingEntries(messagesToRead);
}
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Schedule read of {} messages or {} bytes", replicatorId, newInFlightTask.readingEntries,
+ bytesToRead);
+ }
+ cursor.asyncReadEntriesOrWait(newInFlightTask.readingEntries, bytesToRead, this,
+ newInFlightTask/* Context object */, topic.getMaxReadPosition());
}
@Override
@@ -338,7 +334,10 @@ public abstract class PersistentReplicator extends AbstractReplicator
if (log.isDebugEnabled()) {
log.debug("[{}] Read entries complete of {} messages", replicatorId, entries.size());
}
+ InFlightTask inFlightTask = (InFlightTask) ctx;
+ inFlightTask.setEntries(entries);
+ // After the replicator starts, the speed will be gradually increased.
int maxReadBatchSize = topic.getBrokerService().pulsar().getConfiguration().getDispatcherMaxReadBatchSize();
if (readBatchSize < maxReadBatchSize) {
int newReadBatchSize = Math.min(readBatchSize * 2, maxReadBatchSize);
@@ -352,9 +351,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
readFailureBackoff.reduceToHalf();
- boolean atLeastOneMessageSentForReplication = replicateEntries(entries);
-
- HAVE_PENDING_READ_UPDATER.set(this, FALSE);
+ boolean atLeastOneMessageSentForReplication = replicateEntries(entries, inFlightTask);
if (atLeastOneMessageSentForReplication && !isWritable()) {
// Don't read any more entries until the current pending entries are persisted
@@ -367,7 +364,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
}
- protected abstract boolean replicateEntries(List<Entry> entries);
+ protected abstract boolean replicateEntries(List<Entry> entries, InFlightTask inFlightTask);
protected CompletableFuture<SchemaInfo> getSchemaInfo(MessageImpl msg) throws ExecutionException {
if (msg.getSchemaVersion() == null || msg.getSchemaVersion().length == 0) {
@@ -391,37 +388,40 @@ public abstract class PersistentReplicator extends AbstractReplicator
private PersistentReplicator replicator;
private Entry entry;
private MessageImpl msg;
+ private InFlightTask inFlightTask;
@Override
public void sendComplete(Exception exception) {
if (exception != null && !(exception instanceof PulsarClientException.InvalidMessageException)) {
- log.error("[{}] Error producing on remote broker", replicator.replicatorId, exception);
- // cursor should be rewinded since it was incremented when readMoreEntries
- replicator.cursor.rewind();
+ log.error("[{}] Error producing on remote broker， in-flight messages: {}, producer pending queue size:"
+ + " {}",
+ replicator.replicatorId, replicator.inFlightTasks,
+ replicator.producer.getPendingQueueSize(), exception);
+ // cursor should be rewound since it was incremented when readMoreEntries
+ replicator.beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Failed_Publishing);
+ replicator.doRewindCursor(false);
} else {
if (log.isDebugEnabled()) {
log.debug("[{}] Message persisted on remote broker", replicator.replicatorId, exception);
}
+ inFlightTask.incCompletedEntries();
replicator.cursor.asyncDelete(entry.getPosition(), replicator, entry.getPosition());
}
entry.release();
- int pending = PENDING_MESSAGES_UPDATER.decrementAndGet(replicator);
-
// In general, we schedule a new batch read operation when the occupied queue size gets smaller than half
// the max size, unless another read operation is already in progress.
// If the producer is not currently writable (disconnected or TCP window full), we want to defer the reads
// until we have emptied the whole queue, and at that point we will read a batch of 1 single message if the
// producer is still not "writable".
- if (pending < replicator.producerQueueThreshold //
- && HAVE_PENDING_READ_UPDATER.get(replicator) == FALSE //
- ) {
- if (pending == 0 || replicator.producer.isWritable()) {
+ int permits = replicator.getPermitsIfNoPendingRead();
+ if (replicator.producerQueueSize - permits < replicator.producerQueueThreshold) {
+ if (replicator.producerQueueSize == permits || replicator.producer.isWritable()) {
replicator.readMoreEntries();
} else {
if (log.isDebugEnabled()) {
log.debug("[{}] Not resuming reads. pending: {} is-writable: {}",
- replicator.replicatorId, pending,
+ replicator.replicatorId, replicator.producerQueueSize - permits,
replicator.producer.isWritable());
}
}
@@ -436,15 +436,18 @@ public abstract class PersistentReplicator extends AbstractReplicator
this.recyclerHandle = recyclerHandle;
}
- static ProducerSendCallback create(PersistentReplicator replicator, Entry entry, MessageImpl msg) {
+ static ProducerSendCallback create(PersistentReplicator replicator, Entry entry, MessageImpl msg,
+ InFlightTask inFlightTask) {
ProducerSendCallback sendCallback = RECYCLER.get();
sendCallback.replicator = replicator;
sendCallback.entry = entry;
sendCallback.msg = msg;
+ sendCallback.inFlightTask = inFlightTask;
return sendCallback;
}
private void recycle() {
+ inFlightTask = null;
replicator = null;
entry = null; //already released and recycled on sendComplete
if (msg != null) {
@@ -514,7 +517,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
}
- HAVE_PENDING_READ_UPDATER.set(this, FALSE);
brokerService.executor().schedule(this::readMoreEntries, waitTimeMillis, TimeUnit.MILLISECONDS);
}
@@ -753,4 +755,210 @@ public abstract class PersistentReplicator extends AbstractReplicator
public ManagedCursor getCursor() {
return cursor;
}
-}
+
+ @Data
+ protected static class InFlightTask {
+ PositionImpl readPos;
+ int readingEntries;
+ volatile List<Entry> entries;
+ volatile int completedEntries;
+ volatile boolean skipReadResultDueToCursorRewind;
+ final String replicatorId;
+
+ public synchronized void incCompletedEntries() {
+ if (!CollectionUtils.isEmpty(entries) && completedEntries < entries.size()) {
+ completedEntries++;
+ } else {
+ log.error("Unexpected calling of increase completed entries. {}", this.toString());
+ }
+ }
+
+ synchronized void recycle(PositionImpl readStart, int readingEntries) {
+ this.readPos = readStart;
+ this.readingEntries = readingEntries;
+ this.entries = null;
+ this.completedEntries = 0;
+ this.skipReadResultDueToCursorRewind = false;
+ }
+
+ public InFlightTask(PositionImpl readPos, int readingEntries, String replicatorId) {
+ this.readPos = readPos;
+ this.readingEntries = readingEntries;
+ this.replicatorId = replicatorId;
+ }
+
+ public boolean isDone() {
+ if (entries == null) {
+ return false;
+ }
+ if (entries != null && entries.isEmpty()) {
+ return true;
+ }
+ return completedEntries >= entries.size();
+ }
+
+ @Override
+ public String toString() {
+ return "Replicator InFlightTask "
+ + "{replicatorId=" + replicatorId
+ + ", readPos=" + readPos
+ + ", readingEntries=" + readingEntries
+ + ", readoutEntries=" + (entries == null ? "-1" : entries.size())
+ + ", completedEntries=" + completedEntries
+ + ", skipReadResultDueToCursorRewound=" + skipReadResultDueToCursorRewind
+ + "}";
+ }
+ }
+
+ @VisibleForTesting
+ InFlightTask createOrRecycleInFlightTaskIntoQueue(PositionImpl readPos, int readingEntries) {
+ synchronized (inFlightTasks) {
+ // Reuse projects that has done.
+ if (!inFlightTasks.isEmpty()) {
+ InFlightTask first = inFlightTasks.peek();
+ if (first.isDone()) {
+ // Remove from the first index, and add to the latest index.
+ inFlightTasks.poll();
+ first.recycle(readPos, readingEntries);
+ inFlightTasks.add(first);
+ return first;
+ }
+ }
+ // New project if nothing can be reused.
+ InFlightTask task = new InFlightTask(readPos, readingEntries, replicatorId);
+ inFlightTasks.add(task);
+ return task;
+ }
+ }
+
+ protected InFlightTask acquirePermitsIfNotFetchingSchema() {
+ synchronized (inFlightTasks) {
+ if (hasPendingRead()) {
+ log.info("[{}] Skip the reading because there is a pending read task", replicatorId);
+ return null;
+ }
+ if (waitForCursorRewindingRefCnf > 0) {
+ log.info("[{}] Skip the reading due to new detected schema", replicatorId);
+ return null;
+ }
+ if (state != Started) {
+ log.info("[{}] Skip the reading because producer has not started [{}]", replicatorId, state);
+ return null;
+ }
+ // Guarantee that there is a unique cursor reading task.
+ int permits = getPermitsIfNoPendingRead();
+ if (permits == 0) {
+ return null;
+ }
+ return createOrRecycleInFlightTaskIntoQueue((PositionImpl) cursor.getReadPosition(), permits);
+ }
+ }
+
+ protected int getPermitsIfNoPendingRead() {
+ synchronized (inFlightTasks) {
+ for (InFlightTask task : inFlightTasks) {
+ boolean hasPendingCursorRead = task.readPos != null && task.entries == null;
+ if (hasPendingCursorRead) {
+ // Skip the current reading if there is a pending cursor reading.
+ return 0;
+ }
+ }
+ return producerQueueSize - getInflightMessagesCount();
+ }
+ }
+
+ protected int getInflightMessagesCount() {
+ int inFlight = 0;
+ synchronized (inFlightTasks) {
+ for (InFlightTask task : inFlightTasks) {
+ if (task.isDone()) {
+ continue;
+ }
+ if (task.entries == null) {
+ inFlight += task.readingEntries;
+ continue;
+ }
+ inFlight += Math.max(task.entries.size() - task.completedEntries, 0);
+ }
+ }
+ return inFlight;
+ }
+
+ protected CompletableFuture<Void> beforeDisconnect() {
+ // Ensure no in-flight task.
+ synchronized (inFlightTasks) {
+ for (PersistentReplicator.InFlightTask task : inFlightTasks) {
+ if (!task.isDone() && task.readPos.compareTo(
+ (PositionImpl) cursor.getManagedLedger().getLastConfirmedEntry()) < 0) {
+ return CompletableFuture.failedFuture(new BrokerServiceException
+ .TopicBusyException("Cannot close a replicator with backlog"));
+ }
+ }
+ beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Disconnecting);
+ return CompletableFuture.completedFuture(null);
+ }
+ }
+
+ protected void afterDisconnected() {
+ doRewindCursor(false);
+ }
+
+ protected void beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding reason) {
+ synchronized (inFlightTasks) {
+ boolean hasCanceledPendingRead = cursor.cancelPendingReadRequest();
+ reasonOfWaitForCursorRewinding = reason;
+ waitForCursorRewindingRefCnf += 1;
+ cancelPendingReadTasks(hasCanceledPendingRead);
+ }
+ }
+
+ protected void doRewindCursor(boolean triggerReadMoreEntries) {
+ synchronized (inFlightTasks) {
+ cursor.rewind();
+ waitForCursorRewindingRefCnf -= 1;
+ reasonOfWaitForCursorRewinding = null;
+ }
+ if (triggerReadMoreEntries) {
+ readMoreEntries();
+ }
+ }
+
+ private void cancelPendingReadTasks(boolean canceledPendingRead) {
+ InFlightTask readingTask = null;
+ synchronized (inFlightTasks) {
+ for (InFlightTask task : inFlightTasks) {
+ task.setSkipReadResultDueToCursorRewind(true);
+ if (task.entries == null) {
+ if (readingTask != null) {
+ log.error("Unexpected state because there are more than one tasks' state is pending read. {}",
+ inFlightTasks);
+ }
+ readingTask = task;
+ }
+ }
+ // Correct state to avoid a replicate stuck because a pending reading task occupies permits.
+ // There is at most one reading task.
+ // The task will never receive a read completed callback if cancel pending reading successfully.
+ if (canceledPendingRead && readingTask != null) {
+ readingTask.setEntries(Collections.emptyList());
+ }
+ }
+ }
+
+ @Override
+ public void beforeTerminate() {
+ beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Terminating);
+ }
+
+ protected boolean hasPendingRead() {
+ synchronized (inFlightTasks) {
+ for (InFlightTask task : inFlightTasks) {
+ if (task.readPos != null && task.entries == null) {
+ // Skip the current reading if there is a pending cursor reading.
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 84390a97c56..c5a76ddf519 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -615,7 +615,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
decrementPendingWriteOpsAndCheck();
break;
default:
- publishContext.completed(new MessageDeduplication.MessageDupUnknownException(), -1, -1);
+ publishContext.completed(
+ new MessageDeduplication.MessageDupUnknownException(
+ topic, publishContext.getProducerName()), -1, -1);
decrementPendingWriteOpsAndCheck();
}
@@ -860,8 +862,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private synchronized CompletableFuture<Void> closeReplProducersIfNoBacklog() {
List<CompletableFuture<Void>> closeFutures = new ArrayList<>();
- replicators.forEach((region, replicator) -> closeFutures.add(replicator.disconnect(true, true)));
- shadowReplicators.forEach((__, replicator) -> closeFutures.add(replicator.disconnect(true, true)));
+ replicators.forEach((region, replicator) -> closeFutures.add(replicator.disconnect()));
+ shadowReplicators.forEach((__, replicator) -> closeFutures.add(replicator.disconnect()));
return FutureUtil.waitForAll(closeFutures);
}
@@ -4063,7 +4065,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
decrementPendingWriteOpsAndCheck();
break;
default:
- publishContext.completed(new MessageDeduplication.MessageDupUnknownException(), -1, -1);
+ publishContext.completed(new MessageDeduplication.MessageDupUnknownException(
+ topic, publishContext.getProducerName()), -1, -1);
decrementPendingWriteOpsAndCheck();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
index d9dd19f6e88..e4f538d27c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
@@ -55,15 +55,23 @@ public class ShadowReplicator extends PersistentReplicator {
}
@Override
- protected boolean replicateEntries(List<Entry> entries) {
+ protected boolean replicateEntries(List<Entry> entries, InFlightTask inFlightTask) {
boolean atLeastOneMessageSentForReplication = false;
try {
// This flag is set to true when we skip at least one local message,
// in order to skip remaining local messages.
boolean isLocalMessageSkippedOnce = false;
+ boolean skipRemainingMessages = inFlightTask.isSkipReadResultDueToCursorRewind();
for (int i = 0; i < entries.size(); i++) {
Entry entry = entries.get(i);
+ // Skip the messages since the replicator need to fetch the schema info to replicate the schema to the
+ // remote cluster. Rewind the cursor first and continue the message read after fetched the schema.
+ if (skipRemainingMessages) {
+ inFlightTask.incCompletedEntries();
+ entry.release();
+ continue;
+ }
int length = entry.getLength();
ByteBuf headersAndPayload = entry.getDataBuffer();
MessageImpl msg;
@@ -73,6 +81,7 @@ public class ShadowReplicator extends PersistentReplicator {
log.error("[{}] Failed to deserialize message at {} (buffer size: {}): {}", replicatorId,
entry.getPosition(), length, t.getMessage(), t);
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
continue;
}
@@ -84,6 +93,7 @@ public class ShadowReplicator extends PersistentReplicator {
replicatorId, entry.getPosition(), msg.getReplicateTo());
}
cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -97,6 +107,7 @@ public class ShadowReplicator extends PersistentReplicator {
replicatorId, entry.getPosition());
}
isLocalMessageSkippedOnce = true;
+ inFlightTask.incCompletedEntries();
entry.release();
msg.recycle();
continue;
@@ -116,8 +127,7 @@ public class ShadowReplicator extends PersistentReplicator {
headersAndPayload.retain();
// Increment pending messages for messages produced locally
- PENDING_MESSAGES_UPDATER.incrementAndGet(this);
- producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg));
+ producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg, inFlightTask));
atLeastOneMessageSentForReplication = true;
}
} catch (Exception e) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
index 3fac36e4d00..532c4b7858b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
@@ -155,6 +155,9 @@ public class AbstractReplicatorTest {
return false;
}
+ @Override
+ protected void beforeTerminate() {}
+
@Override
public long getNumberOfEntriesInBacklog() {
return 0;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceInternalMethodInvoker.java
similarity index 50%
copy from pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java
copy to pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceInternalMethodInvoker.java
index 5c314397da8..f33642ab13a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceInternalMethodInvoker.java
@@ -18,39 +18,9 @@
*/
package org.apache.pulsar.broker.service;
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
-import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl;
+public class BrokerServiceInternalMethodInvoker {
-public interface Replicator {
-
- void startProducer();
-
- ReplicatorStatsImpl getStats();
-
- CompletableFuture<Void> terminate();
-
- CompletableFuture<Void> disconnect(boolean failIfHasBacklog, boolean closeTheStartingProducer);
-
- void updateRates();
-
- String getRemoteCluster();
-
- default void initializeDispatchRateLimiterIfNeeded() {
- //No-op
+ public static void replicatorSetState(AbstractReplicator replicator, AbstractReplicator.State state) {
+ replicator.state = state;
}
-
- default void updateRateLimiter() {
- }
-
- default Optional<DispatchRateLimiter> getRateLimiter() {
- return Optional.empty();
- }
-
- boolean isConnected();
-
- long getNumberOfEntriesInBacklog();
-
- boolean isTerminated();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index faefda67c73..f982ec998ce 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service;
+import static org.apache.pulsar.broker.service.persistent.BrokerServicePersistInternalMethodInvoker.ensureNoBacklogByInflightTask;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -32,6 +33,7 @@ import static org.testng.Assert.fail;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
+import io.netty.channel.Channel;
import io.netty.util.concurrent.FastThreadLocalThread;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
@@ -70,11 +72,13 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
+import org.apache.pulsar.broker.service.persistent.BrokerServicePersistInternalMethodInvoker;
import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
@@ -82,10 +86,13 @@ import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.impl.ClientBuilderImpl;
+import org.apache.pulsar.client.impl.ClientCnx;
import org.apache.pulsar.client.impl.ProducerBuilderImpl;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.pulsar.common.api.proto.CommandSendReceipt;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
@@ -1338,10 +1345,10 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
}
// Verify: no individual acks.
Awaitility.await().untilAsserted(() -> {
- PersistentTopic persistentTopic2 =
- (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
- assertTrue(
- persistentTopic2.getSubscription(subscriptionName).getNumberOfEntriesInBacklog(true) > 0);
+ PersistentTopic persistentTopic2 =
+ (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
+ assertTrue(
+ persistentTopic2.getSubscription(subscriptionName).getNumberOfEntriesInBacklog(true) > 0);
PersistentTopic persistentTopic1 =
(PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic1.getManagedLedger();
@@ -1484,4 +1491,142 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
assertEquals(valueLocal, publishRateAddLocal2);
});
}
+
+ @Test
+ public void testConcurrencyReplicationReadEntries() throws Exception {
+ String originalReplicationStartAt = pulsar1.getConfig().getReplicationStartAt();
+ int originalDispatcherMaxReadBatchSize = pulsar1.getConfig().getDispatcherMaxReadBatchSize();
+ int originalDispatcherMinReadBatchSize = pulsar1.getConfig().getDispatcherMinReadBatchSize();
+ int originalReplicationProducerQueueSize = pulsar1.getConfig().getReplicationProducerQueueSize();
+ admin1.brokers().updateDynamicConfiguration("replicationStartAt", "earliest");
+ admin1.brokers().updateDynamicConfiguration("dispatcherMaxReadBatchSize", "10");
+ admin1.brokers().updateDynamicConfiguration("dispatcherMinReadBatchSize", "10");
+ admin1.brokers().updateDynamicConfiguration("replicationProducerQueueSize", "10");
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(pulsar1.getConfig().getReplicationStartAt(), "earliest");
+ assertEquals(pulsar1.getConfig().getDispatcherMaxReadBatchSize(), 10);
+ assertEquals(pulsar1.getConfig().getDispatcherMinReadBatchSize(), 10);
+ assertEquals(pulsar1.getConfig().getReplicationProducerQueueSize(), 10);
+ });
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ final String subscriptionName = "s1";
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin1.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+ admin2.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+
+ // Publish messages.
+ Producer<byte[]> producer1 = client1.newProducer().topic(topicName).enableBatching(false).create();
+ CompletableFuture<MessageId> latestSend = null;
+ for (int i = 0; i < 1000; i++) {
+ latestSend = producer1.sendAsync(new byte[]{1});
+ }
+ latestSend.join();
+ log.info("Cluster: {}, Publish finished", cluster1);
+
+ // Inject two delay:
+ // 1. delay publish responding,
+ // 2. delay switch concurrent mechanism of "replicator.readMoreEntries".
+ ClientBuilderImpl clientBuilder2 = (ClientBuilderImpl) PulsarClient.builder().serviceUrl(url2.toString());
+ PulsarClient injectedReplClient2 = InjectedClientCnxClientBuilder.create(clientBuilder2,
+ (conf, eventLoopGroup) -> {
+ return new ClientCnx(conf, eventLoopGroup) {
+
+ @Override
+ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
+ ctx().executor().schedule(() -> super.handleSendReceipt(sendReceipt), 3600, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected Channel channel() {
+ boolean delay = false;
+ StackTraceElement[] stacks = Thread.currentThread().getStackTrace();
+ for (StackTraceElement stack : stacks) {
+ if (stack.toString().contains("readMoreEntries")) {
+ delay = true;
+ break;
+ }
+ }
+ if (!delay) {
+ return super.ctx().channel();
+ }
+ try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return super.ctx().channel();
+ }
+ };
+ });
+ PulsarClient originalReplClient2 = pulsar1.getBrokerService().getReplicationClients()
+ .put(cluster2, injectedReplClient2);
+
+ // Start replication and inject race conditions of "replicator.readMoreEntries".
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ PersistentTopic persistentTopic1 =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+ waitReplicatorStarted(topicName);
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic1.getReplicators().values().iterator().next();
+ for (int i = 0; i < 10; i++) {
+ new Thread(() -> {
+ BrokerServicePersistInternalMethodInvoker.replicatorReadMoreEntries(replicator);
+ }).start();
+ }
+
+ // Verify: after a few seconds, there is no "pending queue is full" error.
+ Thread.sleep(10_000);
+ assertEquals(replicator.producer.getPendingQueueFullCount(), 0);
+
+ // cleanup.
+ producer1.close();
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1));
+ waitReplicatorStopped(topicName);
+ admin1.topics().delete(topicName, false);
+ if (originalReplClient2 == null) {
+ pulsar1.getBrokerService().getReplicationClients().remove(cluster2);
+ } else {
+ pulsar1.getBrokerService().getReplicationClients().put(cluster2, originalReplClient2);
+ }
+ injectedReplClient2.close();
+ admin1.brokers().updateDynamicConfiguration("replicationStartAt", originalReplicationStartAt);
+ admin1.brokers().updateDynamicConfiguration("dispatcherMaxReadBatchSize",
+ originalDispatcherMaxReadBatchSize + "");
+ admin1.brokers().updateDynamicConfiguration("dispatcherMinReadBatchSize",
+ originalDispatcherMinReadBatchSize + "");
+ admin1.brokers().updateDynamicConfiguration("replicationProducerQueueSize",
+ originalReplicationProducerQueueSize + "");
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(pulsar1.getConfig().getReplicationStartAt(), originalReplicationStartAt);
+ assertEquals(pulsar1.getConfig().getDispatcherMaxReadBatchSize(), originalDispatcherMaxReadBatchSize);
+ assertEquals(pulsar1.getConfig().getDispatcherMinReadBatchSize(), originalDispatcherMinReadBatchSize);
+ assertEquals(pulsar1.getConfig().getReplicationProducerQueueSize(), originalReplicationProducerQueueSize);
+ });
+ }
+
+ @Test
+ public void testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ final String subscriptionName = "s1";
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin1.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+
+ // Publish messages.
+ Producer<byte[]> producer1 = client1.newProducer().topic(topicName).enableBatching(false).create();
+ CompletableFuture<MessageId> latestSend = null;
+ for (int i = 0; i < 100; i++) {
+ latestSend = producer1.sendAsync(new byte[]{1});
+ }
+ latestSend.join();
+ log.info("Cluster: {}, Publish finished", cluster1);
+ producer1.close();
+
+ // Start replication.
+ waitForReplicationTaskFinish(topicName);
+ // Verify: all inflight tasks are done.
+ ensureNoBacklogByInflightTask(getReplicator(topicName));
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
index 53e007979dc..a64cb02df1c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
@@ -42,6 +42,9 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.ClientBuilder;
@@ -517,7 +520,7 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
Optional<PartitionedTopicMetadata> metadata = pulsar1.getPulsarResources().getNamespaceResources()
.getPartitionedTopicResources()
.getPartitionedTopicMetadataAsync(topicNameObj).join();
- Function<Replicator, Boolean> ensureNoBacklog = new Function<Replicator,Boolean>() {
+ Function<Replicator, Boolean> ensureNoBacklog = new Function<Replicator, Boolean>() {
@Override
public Boolean apply(Replicator replicator) {
@@ -544,8 +547,27 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
PersistentTopic persistentTopic =
(PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
persistentTopic.getReplicators().values().forEach(replicator -> {
- assertTrue(ensureNoBacklog.apply(replicator));
+ assertTrue(ensureNoBacklog.apply(replicator));
});
}
}
+
+ protected void waitForReplicationTaskFinish(String topicName) throws Exception {
+ PersistentTopic persistentTopic1 = (PersistentTopic) pulsar1.getBrokerService()
+ .getTopic(topicName, false).join().get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic1.getManagedLedger();
+ PositionImpl lac = (PositionImpl) ml.getLastConfirmedEntry();
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().get("pulsar.repl.r2");
+ Awaitility.await().untilAsserted(() -> {
+ if (cursor.getName().startsWith("pulsar.repl")) {
+ assertTrue(((PositionImpl) cursor.getMarkDeletedPosition()).compareTo(lac) >= 0);
+ }
+ });
+ }
+
+ protected GeoPersistentReplicator getReplicator(String topic) {
+ waitReplicatorStarted(topic);
+ return (GeoPersistentReplicator) pulsar1.getBrokerService().getTopic(topic, false).join().get()
+ .getReplicators().get(cluster2);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
index 46047eb3270..a382cb276ac 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
@@ -275,4 +275,16 @@ public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicator
public void testTopicPoliciesReplicationRule() throws Exception {
super.testTopicPoliciesReplicationRule();
}
+
+ @Override
+ @Test
+ public void testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished() throws Exception {
+ super.testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished();
+ }
+
+ @Override
+ @Test(enabled = false)
+ public void testConcurrencyReplicationReadEntries() throws Exception {
+ super.testConcurrencyReplicationReadEntries();
+ }
}
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 80ce679e0e7..a2184176aa1 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
@@ -232,9 +232,21 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
super.testIncompatibleMultiVersionSchema(enableDeduplication);
}
- @Override
+
@Test
public void testTopicPoliciesReplicationRule() throws Exception {
super.testTopicPoliciesReplicationRule();
}
+
+ @Override
+ @Test
+ public void testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished() throws Exception {
+ super.testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished();
+ }
+
+ @Override
+ @Test(enabled = false)
+ public void testConcurrencyReplicationReadEntries() throws Exception {
+ super.testConcurrencyReplicationReadEntries();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
index 1ea8637e9f8..f038fee7504 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
@@ -22,13 +22,11 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Sets;
import java.lang.reflect.Field;
-import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
@@ -37,12 +35,12 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionsController;
import org.apache.pulsar.broker.transaction.buffer.impl.TopicTransactionBufferState;
@@ -798,15 +796,17 @@ public class ReplicatorSubscriptionTest extends ReplicatorTestBase {
}
// unsubscribe replicated subscription in r2
+ admin1.topics().unload(topicName);
admin2.topics().deleteSubscription(topicName, subscriptionName);
final PersistentTopic topic2 = (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
- assertNull(topic2.getSubscription(subscriptionName));
// close replicator producer in r2
- final Method closeReplProducersIfNoBacklog = PersistentTopic.class.getDeclaredMethod("closeReplProducersIfNoBacklog", null);
- closeReplProducersIfNoBacklog.setAccessible(true);
- ((CompletableFuture<Void>) closeReplProducersIfNoBacklog.invoke(topic2, null)).join();
- assertFalse(topic2.getReplicators().get("r1").isConnected());
+ GeoPersistentReplicator replicator2 = (GeoPersistentReplicator) topic2.getReplicators().get("r1");
+ Awaitility.await().untilAsserted(() -> {
+ replicator2.disconnect().join();
+ assertEquals(replicator2.getState(), AbstractReplicator.State.Disconnected);
+ assertFalse(topic2.getReplicators().get("r1").isConnected());
+ });
// send messages in r1
int numMessages = 6;
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 e101c2a7667..75c418953e3 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
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service;
import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName;
+import static org.apache.pulsar.broker.service.persistent.BrokerServicePersistInternalMethodInvoker.ensureNoBacklogByInflightTask;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
@@ -106,7 +107,6 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.schema.Schemas;
import org.awaitility.Awaitility;
-import org.awaitility.reflect.WhiteboxImpl;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -461,9 +461,7 @@ public class ReplicatorTest extends ReplicatorTestBase {
// Verify "pendingMessages" still is correct even if error occurs.
PersistentReplicator replicator = ensureReplicatorCreated(topic, pulsar1);
waitReplicateFinish(topic, admin1);
- Awaitility.await().untilAsserted(() -> {
- assertEquals((int) WhiteboxImpl.getInternalState(replicator, "pendingMessages"), 0);
- });
+ ensureNoBacklogByInflightTask(replicator);
}
@Test
@@ -1735,11 +1733,11 @@ public class ReplicatorTest extends ReplicatorTestBase {
Producer<byte[]> persistentProducer1 = client1.newProducer().topic(topic.toString()).create();
// Send V1 message, which will be replicated to the remote cluster by the replicator.
persistentProducer1.send("V1".getBytes());
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar1.getBrokerService().getTopicReference(topic.toString()).get();
waitReplicateFinish(topic, admin1);
// Pause replicator
- PersistentTopic persistentTopic =
- (PersistentTopic) pulsar1.getBrokerService().getTopicReference(topic.toString()).get();
persistentTopic.getReplicators().forEach((cluster, replicator) -> {
PersistentReplicator persistentReplicator = (PersistentReplicator) replicator;
pauseReplicator(persistentReplicator);
@@ -1754,7 +1752,15 @@ public class ReplicatorTest extends ReplicatorTestBase {
// Start replicator
persistentTopic.getReplicators().forEach((cluster, replicator) -> {
PersistentReplicator persistentReplicator = (PersistentReplicator) replicator;
- persistentReplicator.startProducer();
+ resumeReplicator(persistentReplicator);
+ Awaitility.await().untilAsserted(() -> {
+ CompletableFuture<Optional<Topic>> topic2 =
+ pulsar2.getBrokerService().getTopic(topic.toString(), false);
+ assertTrue(topic2 != null && topic2.isDone() && topic2.get().isPresent());
+ assertEquals(persistentReplicator.getState(), AbstractReplicator.State.Started);
+ });
+ assertEquals(persistentReplicator.getState(), AbstractReplicator.State.Started);
+ ensureNoBacklogByInflightTask(persistentReplicator);
});
waitReplicateFinish(topic, admin1);
@@ -1842,6 +1848,13 @@ public class ReplicatorTest extends ReplicatorTestBase {
Awaitility.await().untilAsserted(() -> {
assertTrue(replicator.isConnected());
});
- replicator.closeProducerAsync(true);
+ Awaitility.await().until(() -> {
+ replicator.disconnect().join();
+ return true;
+ });
+ }
+
+ private void resumeReplicator(PersistentReplicator replicator) {
+ replicator.startProducer();
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BrokerServicePersistInternalMethodInvoker.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BrokerServicePersistInternalMethodInvoker.java
new file mode 100644
index 00000000000..03c092180eb
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BrokerServicePersistInternalMethodInvoker.java
@@ -0,0 +1,47 @@
+/*
+ * 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.service.persistent;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.awaitility.Awaitility;
+
+public class BrokerServicePersistInternalMethodInvoker {
+
+ public static void replicatorReadMoreEntries(GeoPersistentReplicator replicator) {
+ replicator.readMoreEntries();
+ }
+
+ public static void ensureNoBacklogByInflightTask(PersistentReplicator replicator) {
+ Awaitility.await().atMost(20, TimeUnit.SECONDS).until(() -> {
+ synchronized (replicator.inFlightTasks) {
+ for (PersistentReplicator.InFlightTask task : replicator.inFlightTasks) {
+ if (task.readPos.compareTo(
+ (PositionImpl) replicator.cursor.getManagedLedger().getLastConfirmedEntry()) >= 0) {
+ continue;
+ }
+ if (!task.isDone()) {
+ return false;
+ }
+ }
+ }
+ return true;
+ });
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
new file mode 100644
index 00000000000..5bb9c57a742
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
@@ -0,0 +1,367 @@
+/*
+ * 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.service.persistent;
+
+import static org.mockito.Mockito.mock;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.AbstractReplicator;
+import org.apache.pulsar.broker.service.BrokerServiceInternalMethodInvoker;
+import org.apache.pulsar.broker.service.OneWayReplicatorTestBase;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator.InFlightTask;
+import org.apache.pulsar.client.api.MessageId;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class PersistentReplicatorInflightTaskTest extends OneWayReplicatorTestBase {
+
+ private final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ private final String subscriptionName = "s1";
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.setup();
+ createTopics();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+
+ private void createTopics() throws Exception {
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin1.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+ }
+
+ @Test
+ public void testCreateOrRecycleInFlightTaskIntoQueue() throws Exception {
+ log.info("Starting testCreateOrRecycleInFlightTaskIntoQueue");
+
+ // Get the replicator for the test topic
+ PersistentReplicator replicator = getReplicator(topicName);
+ Assert.assertNotNull(replicator, "Replicator should not be null");
+
+ // Get access to the inFlightTasks list for verification
+ LinkedList<InFlightTask> inFlightTasks = replicator.inFlightTasks;
+ Assert.assertNotNull(inFlightTasks, "InFlightTasks list should not be null");
+
+ // Clear any existing tasks to start with a clean state
+ List<InFlightTask> originalTasks = new ArrayList<>(inFlightTasks);
+ inFlightTasks.clear();
+
+ // Test Case 1: Create a new task when the queue is empty
+ PositionImpl position1 = new PositionImpl(1, 1);
+ Assert.assertNotNull(position1, "Position should not be null");
+ InFlightTask task1 = replicator.createOrRecycleInFlightTaskIntoQueue(position1, 10);
+ // Verify a new task was created and added to the queue
+ Assert.assertNotNull(task1, "Task should not be null");
+ Assert.assertEquals(inFlightTasks.size(), 1, "Queue should have one task");
+ Assert.assertEquals(task1.getReadPos(), position1, "Task should have the correct position");
+ Assert.assertEquals(task1.getReadingEntries(), 10, "Task should have the correct reading entries count");
+ // Mark the task as done to test recycling
+ task1.setEntries(Collections.emptyList());
+
+ // Test Case 2: Recycle an existing task
+ PositionImpl position2 = new PositionImpl(2, 2);
+ Assert.assertNotNull(position2, "Position should not be null");
+ InFlightTask task2 = replicator.createOrRecycleInFlightTaskIntoQueue(position2, 20);
+ // Verify the task was recycled
+ Assert.assertNotNull(task2, "Task should not be null");
+ Assert.assertEquals(inFlightTasks.size(), 1, "Queue should still have one task");
+ Assert.assertEquals(task2.getReadPos(), position2, "Task should have the updated position");
+ Assert.assertEquals(task2.getReadingEntries(), 20, "Task should have the updated reading entries count");
+
+ // Test Case 3: Create a new task when no tasks can be recycled
+ task2.setEntries(null); // Make the task not done
+ PositionImpl position3 = new PositionImpl(3, 3);
+ Assert.assertNotNull(position3, "Position should not be null");
+ InFlightTask task3 = replicator.createOrRecycleInFlightTaskIntoQueue(position3, 30);
+ // Verify a new task was created
+ Assert.assertNotNull(task3, "Task should not be null");
+ Assert.assertEquals(inFlightTasks.size(), 2, "Queue should have two tasks");
+ Assert.assertEquals(task3.getReadPos(), position3, "Task should have the correct position");
+ Assert.assertEquals(task3.getReadingEntries(), 30, "Task should have the correct reading entries count");
+
+ // cleanup.
+ log.info("Completed testCreateOrRecycleInFlightTaskIntoQueue");
+ inFlightTasks.clear();
+ inFlightTasks.addAll(originalTasks);
+ }
+
+ @Test
+ public void testGetInflightMessagesCount() throws Exception {
+ log.info("Starting testGetInflightMessagesCount");
+
+ // Get the replicator for the test topic
+ PersistentReplicator replicator = getReplicator(topicName);
+ Assert.assertNotNull(replicator, "Replicator should not be null");
+
+ // Get access to the inFlightTasks list for setup
+ LinkedList<InFlightTask> inFlightTasks = replicator.inFlightTasks;
+ Assert.assertNotNull(inFlightTasks, "InFlightTasks list should not be null");
+
+ // Save original tasks and clear for testing
+ List<InFlightTask> originalTasks = new ArrayList<>(inFlightTasks);
+ inFlightTasks.clear();
+
+ try {
+ // Test Case 1: no task.
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 0);
+
+ // Test Case 2: cursor reading.
+ PositionImpl position1 = new PositionImpl(1, 1);
+ InFlightTask task1 = new InFlightTask(position1, 3, "");
+ inFlightTasks.add(task1);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 3);
+
+ // Test Case 3: read completed.
+ inFlightTasks.clear();
+ PositionImpl position2 = new PositionImpl(2, 2);
+ InFlightTask task2 = new InFlightTask(position2, 3, "");
+ task2.setEntries(Arrays.asList(mock(Entry.class), mock(Entry.class)));
+ inFlightTasks.add(task2);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 2);
+
+ // Test Case 4: Task with some completed entries
+ task2.setCompletedEntries(1);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 1);
+
+ // Test Case 5: Task with all entries completed
+ task2.setCompletedEntries(2);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 0);
+
+ // Test Case 6: Multiple tasks with different states
+ // task2 has 0 in-flight (2 completed out of 2)
+ // task3 has 2 in-flight (1 completed out of 3)
+ PositionImpl position3 = new PositionImpl(3, 3);
+ InFlightTask task3 = new InFlightTask(position3, 4, "");
+ task3.setEntries(Arrays.asList(mock(Entry.class), mock(Entry.class), mock(Entry.class)));
+ task3.setCompletedEntries(1);
+ inFlightTasks.add(task3);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 2);
+
+ // Test Case 7: Multiple tasks with different states
+ // task2 has 0 in-flight (2 completed out of 2)
+ // task3 has 2 in-flight (1 completed out of 3)
+ // task4 has 0 in-flight (empty readoutEntries)
+ PositionImpl position4 = new PositionImpl(4, 4);
+ InFlightTask task4 = new InFlightTask(position4, 2, "");
+ task4.setEntries(Collections.emptyList());
+ inFlightTasks.add(task4);
+ Assert.assertEquals(replicator.getInflightMessagesCount(), 2);
+
+ log.info("Completed testGetInflightMessagesCount");
+ } finally {
+ // Restore original tasks
+ inFlightTasks.clear();
+ inFlightTasks.addAll(originalTasks);
+ }
+ }
+
+ @Test
+ public void testGetPermitsIfNoPendingRead() throws Exception {
+ log.info("Starting testGetPermitsIfNoPendingRead");
+
+ // Get the replicator for the test topic
+ PersistentReplicator replicator = getReplicator(topicName);
+ Assert.assertNotNull(replicator, "Replicator should not be null");
+
+ // Get access to the inFlightTasks list for setup
+ LinkedList<InFlightTask> inFlightTasks = replicator.inFlightTasks;
+ Assert.assertNotNull(inFlightTasks, "InFlightTasks list should not be null");
+
+ // Save original tasks and clear for testing
+ List<InFlightTask> originalTasks = new ArrayList<>(inFlightTasks);
+ inFlightTasks.clear();
+
+ try {
+ // Test Case 1: Empty queue - should return producerQueueSize (1000)
+ Assert.assertEquals(replicator.getPermitsIfNoPendingRead(), 1000,
+ "With empty queue, should return full producerQueueSize");
+
+ // Test Case 2: Task with pending read (readPos != null && readoutEntries == null)
+ PositionImpl position1 = new PositionImpl(1, 1);
+ InFlightTask pendingReadTask = new InFlightTask(position1, 5, "");
+ // Don't set readoutEntries to simulate pending read
+ inFlightTasks.add(pendingReadTask);
+ Assert.assertEquals(replicator.getPermitsIfNoPendingRead(), 0,
+ "With pending read task, should return 0");
+
+ // Test Case 3: Task with completed read but in-flight messages
+ inFlightTasks.clear();
+ PositionImpl position2 = new PositionImpl(2, 2);
+ InFlightTask completedReadTask = new InFlightTask(position2, 5, "");
+ completedReadTask.setEntries(Arrays.asList(
+ mock(Entry.class), mock(Entry.class), mock(Entry.class)));
+ inFlightTasks.add(completedReadTask);
+ Assert.assertEquals(replicator.getPermitsIfNoPendingRead(), 1000 - 3,
+ "With completed read task, should return producerQueueSize - inflightMessages");
+
+ // Test Case 4: Multiple tasks with no pending reads
+ PositionImpl position3 = new PositionImpl(3, 3);
+ InFlightTask task2 = new InFlightTask(position3, 5, "");
+ task2.setEntries(Arrays.asList(mock(Entry.class), mock(Entry.class)));
+ task2.setCompletedEntries(1); // 1 in-flight message
+ inFlightTasks.add(task2);
+ // Now we have 3 + 1 = 4 in-flight messages
+ Assert.assertEquals(replicator.getPermitsIfNoPendingRead(), 1000 - 4,
+ "With multiple tasks, should return producerQueueSize - total inflightMessages");
+
+ // Test Case 5: Multiple tasks including one with pending read
+ PositionImpl position4 = new PositionImpl(4, 4);
+ InFlightTask pendingReadTask2 = new InFlightTask(position4, 5, "");
+ // Don't set readoutEntries to simulate pending read
+ inFlightTasks.add(pendingReadTask2);
+ Assert.assertEquals(replicator.getPermitsIfNoPendingRead(), 0,
+ "With any pending read task, should return 0 regardless of other tasks");
+
+ log.info("Completed testGetPermitsIfNoPendingRead");
+ } finally {
+ // Restore original tasks
+ inFlightTasks.clear();
+ inFlightTasks.addAll(originalTasks);
+ }
+ }
+
+ @Test
+ public void testAcquirePermitsIfNotFetchingSchema() throws Exception {
+ log.info("Starting testAcquirePermitsIfNotFetchingSchema");
+ // Get the replicator for the test topic
+ PersistentReplicator replicator = getReplicator(topicName);
+ Assert.assertNotNull(replicator, "Replicator should not be null");
+
+ // Get access to the inFlightTasks list for setup
+ LinkedList<InFlightTask> inFlightTasks = replicator.inFlightTasks;
+ Assert.assertNotNull(inFlightTasks, "InFlightTasks list should not be null");
+
+ // Save original tasks and clear for testing
+ List<InFlightTask> originalTasks = new ArrayList<>(inFlightTasks);
+ inFlightTasks.clear();
+
+ // Save original state
+ int originalWaitForCursorRewinding = replicator.waitForCursorRewindingRefCnf;
+ AbstractReplicator.State originalState = replicator.getState();
+
+ try {
+ // Test Case 1: Normal case - no pending read, not waiting for cursor rewinding, state is Started
+ // Should return a new InFlightTask
+ // First, check the current permits available
+ int expectedPermits = replicator.getPermitsIfNoPendingRead();
+ Assert.assertTrue(expectedPermits > 0, "Should have available permits for the test");
+ InFlightTask task1 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNotNull(task1, "Should return a new InFlightTask in normal case");
+ Assert.assertNotNull(task1.getReadPos(), "Task should have a read position");
+ Assert.assertEquals(task1.getReadingEntries(), expectedPermits,
+ "Task readingEntries should equal the number of permits available");
+ Assert.assertTrue(inFlightTasks.contains(task1),
+ "Task should be added to the inFlightTasks list");
+
+ // Test Case 2: With pending read - should return null
+ inFlightTasks.clear();
+ PositionImpl position1 = new PositionImpl(1, 1);
+ InFlightTask pendingReadTask = new InFlightTask(position1, 5, "");
+ // Don't set readoutEntries to simulate pending read
+ inFlightTasks.add(pendingReadTask);
+ InFlightTask task2 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNull(task2, "Should return null when there is a pending read");
+
+ // Test Case 3: With waitForCursorRewinding=true - should return null
+ inFlightTasks.clear();
+ replicator.waitForCursorRewindingRefCnf = 1;
+ InFlightTask task3 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNull(task3, "Should return null when waiting for cursor rewinding");
+ // Reset for next test
+ replicator.waitForCursorRewindingRefCnf = 0;
+
+ // Test Case 4: With state != Started - should return null
+ // We need to use reflection to modify the state since it's protected by AtomicReferenceFieldUpdater
+ BrokerServiceInternalMethodInvoker.replicatorSetState(replicator, AbstractReplicator.State.Starting);
+ InFlightTask task4 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNull(task4, "Should return null when state is not Started");
+ // Reset state for next test
+ BrokerServiceInternalMethodInvoker.replicatorSetState(replicator, AbstractReplicator.State.Started);
+
+ // Test Case 5: With limited permits - verify readingEntries is set correctly
+ inFlightTasks.clear();
+ // Add a task with some in-flight messages to reduce available permits
+ PositionImpl positionLimited = new PositionImpl(10, 10);
+ InFlightTask limitedTask = new InFlightTask(positionLimited, 5, "");
+ // Add enough entries to leave just a small number of permits (e.g., 10)
+ List<Entry> limitedEntries = new ArrayList<>();
+ int entriesCount = 990;
+ for (int j = 0; j < entriesCount; j++) {
+ limitedEntries.add(mock(Entry.class));
+ }
+ limitedTask.setEntries(limitedEntries);
+ inFlightTasks.add(limitedTask);
+ // Check that we have limited permits available
+ int limitedPermits = replicator.getPermitsIfNoPendingRead();
+ Assert.assertTrue(limitedPermits > 0 && limitedPermits < 20,
+ "Should have a small number of permits available for testing");
+ // Now acquire permits and verify readingEntries matches the limited permits
+ InFlightTask task5 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNotNull(task5, "Should return a task with limited permits");
+ Assert.assertEquals(task5.getReadingEntries(), limitedPermits,
+ "Task readingEntries should equal the limited number of permits available");
+
+ // Test Case 6: With permits=0 - should return null
+ inFlightTasks.clear();
+ // Add tasks that will make getPermitsIfNoPendingRead() return 0
+ // We need enough in-flight messages to equal producerQueueSize
+ for (int i = 0; i < 10; i++) {
+ PositionImpl position = new PositionImpl(i, i);
+ InFlightTask task = new InFlightTask(position, 5, "");
+ List<Entry> entries = new ArrayList<>();
+ for (int j = 0; j < 100; j++) {
+ entries.add(mock(Entry.class));
+ }
+ task.setEntries(entries);
+ inFlightTasks.add(task);
+ }
+ InFlightTask task6 = replicator.acquirePermitsIfNotFetchingSchema();
+ Assert.assertNull(task6, "Should return null when permits is 0");
+ log.info("Completed testAcquirePermitsIfNotFetchingSchema");
+ } finally {
+ // Restore original state
+ replicator.waitForCursorRewindingRefCnf = originalWaitForCursorRewinding;
+ BrokerServiceInternalMethodInvoker.replicatorSetState(replicator, originalState);
+ // Restore original tasks
+ inFlightTasks.clear();
+ inFlightTasks.addAll(originalTasks);
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
index 9f1885e034d..44f02dac81d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
@@ -18,7 +18,7 @@
*/
package org.apache.pulsar.broker.service.persistent;
-import static org.testng.Assert.assertEquals;
+import static org.apache.pulsar.broker.service.persistent.BrokerServicePersistInternalMethodInvoker.ensureNoBacklogByInflightTask;
import static org.testng.Assert.assertTrue;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -43,7 +43,6 @@ import org.apache.pulsar.common.policies.data.ReplicatorStats;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.schema.Schemas;
import org.awaitility.Awaitility;
-import org.awaitility.reflect.WhiteboxImpl;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -123,7 +122,7 @@ public class ShadowReplicatorTest extends BrokerTestBase {
replicator.msgOut.calculateRate();
return replicator.msgOut.getCount() >= 1;
});
- Awaitility.await().until(() -> PersistentReplicator.PENDING_MESSAGES_UPDATER.get(replicator) == 0);
+ ensureNoBacklogByInflightTask(replicator);
PersistentTopic shadowTopic =
(PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopicName).get().get();
@@ -134,7 +133,9 @@ public class ShadowReplicatorTest extends BrokerTestBase {
Assert.assertEquals(shadowMessage.getData(), sourceMessage.getData());
Assert.assertEquals(shadowMessage.getSequenceId(), sourceMessage.getSequenceId());
Assert.assertEquals(shadowMessage.getEventTime(), sourceMessage.getEventTime());
- Assert.assertEquals(shadowMessage.getProperties(), sourceMessage.getProperties());
+ // Since branch-3.0 does not contain https://github.com/apache/pulsar/pull/23236, this test is not the same as
+ // master-branch.
+ Assert.assertEquals(shadowMessage.getProperties().get("PK"), "PV");
Assert.assertEquals(shadowMessage.getKey(), sourceMessage.getKey());
Assert.assertEquals(shadowMessage.getOrderingKey(), sourceMessage.getOrderingKey());
Assert.assertEquals(shadowMessage.getSchemaVersion(), sourceMessage.getSchemaVersion());
@@ -164,7 +165,7 @@ public class ShadowReplicatorTest extends BrokerTestBase {
}
@Test
- public void testCounterOfPengdingMessagesCorrect() throws Exception {
+ public void testCounterOfPendingMessagesCorrect() throws Exception {
TopicName sourceTopicName = TopicName
.get(BrokerTestUtil.newUniqueName("persistent://prop1/ns-source/source-topic"));
TopicName shadowTopicName = TopicName
@@ -191,11 +192,11 @@ public class ShadowReplicatorTest extends BrokerTestBase {
producer.send(new Schemas.PersonOne(i));
}
- // Verify "pendingMessages" still is correct even if error occurs.
+ // Verify "inflight replication tasks" are correct.
PersistentReplicator replicator = getAnyShadowReplicator(sourceTopicName, pulsar);
waitReplicateFinish(sourceTopicName, admin);
- Awaitility.await().untilAsserted(() -> {
- assertEquals((int) WhiteboxImpl.getInternalState(replicator, "pendingMessages"), 0);
- });
+ ensureNoBacklogByInflightTask(replicator);
}
+
+
}
\ No newline at end of file
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 4764ed10fe3..00ecd3d6263 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -464,9 +464,13 @@ public class ClientCnx extends PulsarHandler {
}
ProducerImpl<?> producer = producers.get(producerId);
if (ledgerId == -1 && entryId == -1) {
- log.warn("{} Message with sequence-id {}-{} published by producer [id:{}, name:{}] has been dropped",
- ctx.channel(), sequenceId, highestSequenceId, producerId,
- producer != null ? producer.getProducerName() : "null");
+ if (producer == null) {
+ log.warn("{} Message with sequence-id {}-{} published by producer [id:{}, name:{}] has been dropped",
+ ctx.channel(), sequenceId, highestSequenceId, producerId, "null");
+ } else {
+ producer.printWarnLogWhenCanNotDetermineDeduplication(ctx.channel(), sequenceId, highestSequenceId);
+ }
+
} else {
if (log.isDebugEnabled()) {
log.debug("{} Got receipt for producer: [id:{}, name:{}] -- sequence-id: {}-{} -- entry-id: {}:{}",
@@ -957,7 +961,8 @@ public class ClientCnx extends PulsarHandler {
return ctx;
}
- Channel channel() {
+ @VisibleForTesting
+ protected Channel channel() {
return ctx.channel();
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/GeoReplicationProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/GeoReplicationProducerImpl.java
index 12a6be07070..2595ce4ba61 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/GeoReplicationProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/GeoReplicationProducerImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import io.netty.channel.Channel;
import io.netty.util.ReferenceCountUtil;
import java.util.List;
import java.util.Optional;
@@ -244,6 +245,13 @@ public class GeoReplicationProducerImpl extends ProducerImpl{
&& Markers.isReplicationMarker(op.msg.getMessageBuilder().getMarkerType());
}
+ @Override
+ public void printWarnLogWhenCanNotDetermineDeduplication(Channel channel, long sourceLId, long sourceEId) {
+ log.warn("[{}] producer [id:{}, name:{}, channel: {}] message with source entry {}-{} published by has been"
+ + " dropped because Broker can not determine whether is duplicate or not",
+ topic, producerId, producerName, channel, sourceLId, sourceEId);
+ }
+
private boolean isReplicationMarker(long highestSeq) {
return Long.MIN_VALUE == highestSeq;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index e4c533ab904..59ee51124f0 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -32,6 +32,7 @@ import static org.apache.pulsar.common.protocol.Commands.readChecksum;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.ChannelPromise;
@@ -177,6 +178,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
private boolean errorState;
private final boolean pauseSendingToPreservePublishOrderOnSchemaRegFailure;
+ // This variable can be exposed as a metrics in the future, a PIP is needed.
+ private final AtomicInteger pendingQueueFullCounter;
public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
CompletableFuture<Producer<T>> producerCreatedFuture, int partitionIndex, Schema<T> schema,
@@ -276,11 +279,17 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
metadata = Collections.unmodifiableMap(new HashMap<>(conf.getProperties()));
}
+ pendingQueueFullCounter = new AtomicInteger();
this.connectionHandler = initConnectionHandler();
setChunkMaxMessageSize();
grabCnx();
}
+ @VisibleForTesting
+ public int getPendingQueueFullCount() {
+ return pendingQueueFullCounter.get();
+ }
+
ConnectionHandler initConnectionHandler() {
return new ConnectionHandler(this,
new BackoffBuilder()
@@ -358,6 +367,13 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
return future;
}
+ public void printWarnLogWhenCanNotDetermineDeduplication(Channel channel, long sequenceId,
+ long highestSequenceId) {
+ log.warn("[{}] producer [id:{}, name:{}, channel: {}] message with sequence-id {}-{} published by has been"
+ + " dropped because Broker can not determine whether is duplicate or not",
+ topic, producerId, producerName, channel, sequenceId, highestSequenceId);
+ }
+
private class DefaultSendMessageCallback implements SendCallback {
CompletableFuture<MessageId> sendFuture;
@@ -1015,6 +1031,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
client.getMemoryLimitController().reserveMemory(payloadSize);
} else {
if (!semaphore.map(Semaphore::tryAcquire).orElse(true)) {
+ pendingQueueFullCounter.incrementAndGet();
callback.sendComplete(new PulsarClientException.ProducerQueueIsFullError(
"Producer send queue is full", sequenceId));
return false;
(pulsar) 01/03: [fix][broker] Fix ack hole in cursor for geo-replication (#20931)
Posted by yu...@apache.org.
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
commit e0cc63a2222341b5f5ab105523ece076d3a5a00d
Author: Masahiro Sakamoto <ma...@lycorp.co.jp>
AuthorDate: Thu Oct 10 19:03:56 2024 +0900
[fix][broker] Fix ack hole in cursor for geo-replication (#20931)
Co-authored-by: Masahiro Sakamoto <ma...@yahoo-corp.jp>
(cherry picked from commit 9f8b4a6d15899c7255c525299e1d6011a09b940e)
---
.../service/persistent/PersistentReplicator.java | 25 +++++++++++++++++++---
1 file changed, 22 insertions(+), 3 deletions(-)
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 9330fa9bd52..81afff2012c 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
@@ -110,6 +110,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
private final ReplicatorStatsImpl stats = new ReplicatorStatsImpl();
protected volatile boolean fetchSchemaInProgress = false;
+ private volatile boolean waitForCursorRewinding = false;
public PersistentReplicator(String localCluster, PersistentTopic localTopic, ManagedCursor cursor,
String remoteCluster, String remoteTopic,
@@ -137,9 +138,15 @@ public abstract class PersistentReplicator extends AbstractReplicator
@Override
protected void setProducerAndTriggerReadEntries(Producer<byte[]> producer) {
- // Rewind the cursor to be sure to read again all non-acked messages sent while restarting.
- cursor.rewind();
- cursor.cancelPendingReadRequest();
+ waitForCursorRewinding = true;
+
+ // Repeat until there are no read operations in progress
+ if (STATE_UPDATER.get(this) == State.Starting && HAVE_PENDING_READ_UPDATER.get(this) == TRUE
+ && !cursor.cancelPendingReadRequest()) {
+ brokerService.getPulsar().getExecutor()
+ .schedule(() -> setProducerAndTriggerReadEntries(producer), 10, TimeUnit.MILLISECONDS);
+ return;
+ }
/**
* 1. Try change state to {@link Started}.
@@ -152,6 +159,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
if (!(producer instanceof ProducerImpl)) {
log.error("[{}] The partitions count between two clusters is not the same, the replicator can not be"
+ " created successfully: {}", replicatorId, state);
+ waitForCursorRewinding = false;
doCloseProducerAsync(producer, () -> {});
throw new ClassCastException(producer.getClass().getName() + " can not be cast to ProducerImpl");
}
@@ -162,6 +170,11 @@ public abstract class PersistentReplicator extends AbstractReplicator
backOff.reset();
// activate cursor: so, entries can be cached.
this.cursor.setActive();
+
+ // Rewind the cursor to be sure to read again all non-acked messages sent while restarting
+ cursor.rewind();
+ waitForCursorRewinding = false;
+
// read entries
readMoreEntries();
} else {
@@ -177,6 +190,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
log.error("[{}] Replicator state is not expected, so close the producer. Replicator state: {}",
replicatorId, changeStateRes.getRight());
}
+ waitForCursorRewinding = false;
// Close the producer if change the state fail.
doCloseProducerAsync(producer, () -> {});
}
@@ -268,6 +282,11 @@ public abstract class PersistentReplicator extends AbstractReplicator
// Schedule read
if (HAVE_PENDING_READ_UPDATER.compareAndSet(this, FALSE, TRUE)) {
+ if (waitForCursorRewinding) {
+ log.info("[{}] Skip the reading because repl producer is starting", replicatorId);
+ HAVE_PENDING_READ_UPDATER.set(this, FALSE);
+ return;
+ }
if (log.isDebugEnabled()) {
log.debug("[{}] Schedule read of {} messages", replicatorId, messagesToRead);
}