[fix][test] Fix flaky OneWayReplicatorDeduplicationTest.testDeduplication

[Praveenkumar76]

Fixes apache#25141

Motivation

OneWayReplicatorDeduplicationTest.testDeduplication is flaky because MessageDeduplication.takeSnapshot is designed to drop concurrent snapshot requests if an existing snapshot operation is already in progress. The test relies on specific positions being persisted, but if a manual trigger is dropped due to this intentional "lossy" design, the test fails with a ConditionTimeoutException while waiting for the state to update.

Modifications

• Modified OneWayReplicatorDeduplicationTest to set brokerDeduplicationSnapshotIntervalSeconds to 1 second in the test configuration.

• By increasing the frequency of the background snapshot monitor, the system is guaranteed to capture and persist the deduplication state frequently enough to satisfy the test's assertions, even if a specific manual trigger request is skipped due to a race condition.

Verifying this change

• Verified that the flaky test now runs consistently without failures.
• Confirmed that the 1-second background interval provides a sufficient "safety net" to persist the state when concurrent manual triggers are dropped.
• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as:

• org.apache.pulsar.broker.service.OneWayReplicatorDeduplicationTest.testDeduplication

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[lhotari]

The root cause is a race condition in MessageDeduplication.takeSnapshot. When multiple snapshot requests occur concurrently, the current implementation uses a compareAndSet guard to allow only one active snapshot. If another request arrives while a snapshot is already in progress, the method immediately returns a completed future, effectively dropping the new request.

The title of the PR should reflect that this is a production code change, not a fix to a flaky test.

[lhotari]

I don't think that MessageDuplication needs to be changed. The test itself is invalid and should be fixed instead. Here's the explanation of MessageDuplication, answer from DeepWiki:

MessageDeduplication.takeSnapshot(Position) — Role and Correctness

Role of the method

The private takeSnapshot(Position) in org.apache.pulsar.broker.service.persistent.MessageDeduplication persists the current deduplication state to a ManagedCursor. The snapshot captures the highest sequence IDs persisted per producer (the highestSequencedPersisted map) and marks the cursor at the supplied Position, which corresponds to the last confirmed entry in the ManagedLedger. This is what allows deduplication state to be recovered correctly after a broker restart or topic unload/reload.

When it's invoked

• Periodically, after snapshotInterval entries have been persisted (counter-driven, from recordMessagePersistedNormal / recordMessagePersistedRepl).
• After replaying the cursor on a deduplication status check, if enough entries were processed.
• After purging inactive producers, to persist the trimmed state.
• From a scheduled task driven by brokerDeduplicationSnapshotFrequencyInSeconds / brokerDeduplicationSnapshotIntervalSeconds, via the public takeSnapshot() wrapper.
• From BacklogQuotaManager when the deduplication cursor is the slowest consumer and needs to be advanced.

Correctness requirements when one is already in progress

Concurrency is gated by an AtomicBoolean snapshotTaking. If a snapshot is requested while another is in flight, the new request is dropped (not queued) and a warning is logged — there is no waiting and no replacement. The contract relies on two things to stay correct under that drop:

1. The snapshot reads highestSequencedPersisted at invocation time and pairs it with the supplied Position (last confirmed entry), so each snapshot is internally consistent.
2. The underlying ManagedCursor.markDelete is monotonic — it never moves the cursor backward — so even though concurrent requests are skipped, missing one is safe: a later snapshot will advance the cursor to a position at least as far as the dropped one would have. Combined with the periodic/scheduled triggers, the cursor is guaranteed to keep moving forward.

The practical implication: callers must not assume their requested snapshot actually ran. The mechanism is designed to be lossy-but-monotonic — fine for periodic persistence, but not something to rely on for "snapshot exactly at this position right now" semantics.

---

Source: DeepWiki query on apache/pulsar

One possible way to address the flaky test problem is to configure brokerDeduplicationSnapshotIntervalSeconds to a low value in the test instead of relying on brokerDeduplicationEntriesInterval. This shouldn't change the intention of the test at all.

[Praveenkumar76]

Thanks for the detailed explanation, @lhotari.

My original change aimed to resolve the ConditionTimeoutException by ensuring the test never missed a snapshot signal, but I now understand that the 'lossy-but-monotonic' design is intentional for production performance.

I agree that fixing the test configuration is the better approach here, I will:

Revert the changes to MessageDeduplication.java.

Update the test to use a lower brokerDeduplicationSnapshotIntervalSeconds as suggested to ensure frequent snapshots during verification.

Update the PR title/description to reflect a test stabilization change.

Pushing the update shortly.

[lhotari]

LGTM

[lhotari]

Thanks for the contribution @Praveenkumar76

[Praveenkumar76]

/pulsarbot rerun