[fix][broker]excessive replication speed leads to error: Producer send queue is full

Author: poorbarcode

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java

@@ -398,12 +398,15 @@ protected CompletableFuture<Void> doCloseProducerAsync(Producer<byte[]> producer
         });
     }
 
+    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;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java

@@ -258,4 +258,9 @@ public long getNumberOfEntriesInBacklog() {
     protected void disableReplicatorRead() {
         // No-op
     }
+
+    @Override
+    protected void beforeTerminate() {
+        // No-op
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java

@@ -81,7 +81,7 @@ protected CompletableFuture<Void> prepareCreateProducer() {
     }
 
     @Override
-    protected boolean replicateEntries(List<Entry> entries) {
+    protected boolean replicateEntries(List<Entry> entries, InFlightTask inFlightTask) {
         boolean atLeastOneMessageSentForReplication = false;
         boolean isEnableReplicatedSubscriptions =
                 brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions();
@@ -90,12 +90,13 @@ protected boolean replicateEntries(List<Entry> entries) {
             // 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.isSkipReadResultDueToCursorRewound();
             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;
                 }
@@ -108,12 +109,14 @@ protected boolean replicateEntries(List<Entry> entries) {
                     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;
@@ -123,6 +126,7 @@ protected boolean replicateEntries(List<Entry> entries) {
                             msg.getMessageBuilder().getTxnidLeastBits());
                     if (topic.isTxnAborted(tx, entry.getPosition())) {
                         cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+                        inFlightTask.incCompletedEntries();
                         entry.release();
                         msg.recycle();
                         continue;
@@ -136,6 +140,7 @@ protected boolean replicateEntries(List<Entry> entries) {
                 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;
@@ -147,6 +152,7 @@ protected boolean replicateEntries(List<Entry> entries) {
                                 entry.getPosition(), msg.getReplicateTo());
                     }
                     cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+                    inFlightTask.incCompletedEntries();
                     entry.release();
                     msg.recycle();
                     continue;
@@ -159,19 +165,21 @@ protected boolean replicateEntries(List<Entry> entries) {
                                 replicatorId, entry.getPosition(), msg.getReplicateTo());
                     }
                     cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
+                    inFlightTask.incCompletedEntries();
                     entry.release();
                     msg.recycle();
                     continue;
                 }
 
                 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;
@@ -184,24 +192,31 @@ protected boolean replicateEntries(List<Entry> entries) {
 
                 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) {
                             log.warn("[{}] Failed to get schema from local cluster, will try in the next loop",
                                     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());
@@ -214,7 +229,6 @@ protected boolean replicateEntries(List<Entry> entries) {
                     stats.incrementMsgOutCounter();
                     stats.incrementBytesOutCounter(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());
                     }