[fix][broker] Fix ack hole in cursor for geo-replication

[massakam]

Motivation

Occasionally there is an ack hole in the cursor for geo-replication. The following is the internal stats for the topic where the problem occurred:

{
  "entriesAddedCounter" : 11000,
  "numberOfEntries" : 6999,
  "totalSize" : 362285,
  "currentLedgerEntries" : 6999,
  "currentLedgerSize" : 362285,
  "lastLedgerCreatedTimestamp" : "2023-07-30T16:17:10.137+09:00",
  "waitingCursorsCount" : 1,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "1687807:6998",
  "state" : "LedgerOpened",
  "ledgers" : [ {
    "ledgerId" : 1687807,
    "entries" : 0,
    "size" : 0,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "cursors" : {
    "pulsar.repl.cluster-a" : {
      "markDeletePosition" : "1687807:6002",
      "readPosition" : "1687807:6999",
      "waitingReadOp" : true,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : 10942,
      "cursorLedger" : 1687808,
      "cursorLedgerLastEntry" : 13,
      "individuallyDeletedMessages" : "[(1687807:6060..1687807:6998]]",
      "lastLedgerSwitchTimestamp" : "2023-07-30T16:17:10.143+09:00",
      "state" : "Open",
      "numberOfEntriesSinceFirstNotAckedMessage" : 997,
      "totalNonContiguousDeletedMessagesRange" : 1,
      "properties" : { }
    }
  },
  "schemaLedgers" : [ ],
  "compactedLedger" : {
    "ledgerId" : -1,
    "entries" : -1,
    "size" : -1,
    "offloaded" : false,
    "underReplicated" : false
  }
}

Also, the following log was printed on the broker server. The ack hole is included in the range where the cursor was rewound.

16:50:11.787 [pulsar-io-24-1] INFO  o.a.b.mledger.impl.ManagedCursorImpl - [massakam/test/persistent/t1-pulsar.repl.cluster-a] Rewind from 1687807:6061 to 1687807:5999

This problem occurred in the following situations:

• There are two replication clusters, cluster-a and cluster-b
• cluster-a usually has no producers or consumers, but once a day a producer connects and publishes messages
• Only consumers are connected to cluster-b
• Retention time is 0

In the above case, the producer for geo-replication on the cluster-a side will be closed after a certain period of time by GC.

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

Line 2626 in ca01447

closeReplProducersIfNoBacklog().thenRun(() -> {

However, at this time, an already triggered operation to read new entries will not be cancelled. This operation will remain pending until new entries are available.

Then 24 hours later the user's producer connects again and publishes messages. This triggers the pending operation and causes the replicator to start reading new entries.

However, since the producer for geo-replication has not yet been restarted, these read entries will be dropped without being acknowledged.

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

Lines 138 to 149 in ca01447

	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
	if (log.isDebugEnabled()) {
	log.debug("[{}] Dropping read message at {} because producer is not ready",
	replicatorId, entry.getPosition());
	}
	isLocalMessageSkippedOnce = true;
	entry.release();
	msg.recycle();
	continue;
	}

On the other hand, since the user's producer is connected, the producer for geo-replication is also restarted and the cursor is rewound. After that, the state of the replicator is changed to Started.

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

Lines 138 to 145 in ca01447

	// Rewind the cursor to be sure to read again all non-acked messages sent while restarting
	cursor.rewind();
	cursor.cancelPendingReadRequest();
	HAVE_PENDING_READ_UPDATER.set(this, FALSE);
	this.producer = (ProducerImpl) producer;
	if (STATE_UPDATER.compareAndSet(this, State.Starting, State.Started)) {

At this time, one of the operations triggered before the cursor is rewound succeeds, causing readPosition to move to "the position next to the successfully read entry". Entries before this position will not be read again. As a result, entries that have been read once but not acknowledged will be left as an ack hole.

In short, a race condition between "cursor rewinding when the producer for geo-replication is restarted" and "an read operation that was triggered the last time geo-replication occurred" is what causes this issue.

Modifications

Add a flag named waitForCursorRewinding to the PersistentReplicator class. Normally this value is false. If this value becomes true, the replicator will no longer call cursor.asyncReadEntriesOrWait.

On the other hand, set waitForCursorRewinding to true at the beginning of the readEntries method that is executed when restarting the producer for geo-replication. Then wait until at least one of the following conditions is met:

• state is no longer Starting
  • This means the replicator has been stopped
• havePendingRead becomes FALSE
  • This means there are no reads in progress
• cursor.cancelPendingReadRequest returns true
  • This means that there was a read in progress, but it was successfully canceled

Then change state to Started, rewind the cursor, and set waitForCursorRewinding back to false. This prevents a read triggered before the cursor has been rewound from advancing the cursor again, leaving an ack hole.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[poorbarcode]

@massakam

Nice catch.

I think this issue is the same as the scenario-3 in the Motivation of the PR #20128

I think the PIP-269 can solve this issue.

[massakam]

@poorbarcode Thank you for the information. This issue seems to apply to scenario-1 instead of scenario-3.

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[lhotari]

@massakam Is this PR still needed? If so, please rebase. Otherwise, please close.

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 46.66667% with 8 lines in your changes missing coverage. Please review.

Project coverage is 74.35%. Comparing base (bbc6224) to head (56f5fd9).
Report is 661 commits behind head on master.

Files with missing lines	Patch %	Lines
...roker/service/persistent/PersistentReplicator.java	46.66%	6 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #20931      +/-   ##
============================================
+ Coverage     73.57%   74.35%   +0.78%     
- Complexity    32624    34892    +2268     
============================================
  Files          1877     1949      +72     
  Lines        139502   146880    +7378     
  Branches      15299    16171     +872     
============================================
+ Hits         102638   109215    +6577     
- Misses        28908    29250     +342     
- Partials       7956     8415     +459     

Flag	Coverage Δ
inttests	27.34% <26.66%> (+2.76%)	⬆️
systests	24.42% <0.00%> (+0.09%)	⬆️
unittests	73.70% <46.66%> (+0.86%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...roker/service/persistent/PersistentReplicator.java	67.72% <46.66%> (-1.16%)	⬇️

... and 623 files with indirect coverage changes

[massakam]

@lhotari

@poorbarcode said that PIP-269 could solve this issue, but unfortunately PIP-269 doesn't seem to be progressing. So this issue is not yet resolved and this PR is still needed. I have resolved the conflict, so please review if possible.

[lhotari]

LGTM, great work @massakam