[improve][broker] Recalculate delayed delivery time using server clock to mitigate clock skew

[Radiancebobo]

Motivation

When clients and brokers have clock skew, delayed message delivery can behave incorrectly. For example:

• If a client's clock is 5 minutes behind the server and uses deliverAfter(3min), the message may be
  delivered immediately instead of after 3 minutes
• If a client's clock is 5 minutes ahead, the message may be delivered 5 minutes later than intended

This happens because the broker uses the client-provided deliverAtTime timestamp directly, which is
calculated based on the client's clock.

Modifications

• Modified PersistentDispatcherMultipleConsumers.trackDelayedDelivery() to recalculate delivery time based on
  server clock
• Modified PersistentDispatcherMultipleConsumersClassic.trackDelayedDelivery() with the same logic
• The fix calculates the relative delay (deliverAtTime - publishTime) and adds it to the current server time
• Falls back to original behavior when publishTime is missing or when relativeDelay <= 0
• Added comprehensive unit tests in TrackDelayedDeliveryClockSkewTest covering various clock skew scenarios

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests TrackDelayedDeliveryClockSkewTest with 7 test cases:
  • testDeliverAfterWithClientClockBehindServer - verifies correct behavior when client clock is behind
  • testDeliverAfterWithClientClockAheadOfServer - verifies correct behavior when client clock is ahead
  • testFallbackWhenNoPublishTime - verifies fallback when publishTime is missing
  • testFallbackWhenRelativeDelayNegative - verifies fallback for abnormal data
  • testNoDeliverAtTime - verifies normal non-delayed messages
  • testNoClockSkew - verifies behavior with synchronized clocks
  • testClassicDispatcherClockSkewRecalculation - verifies Classic dispatcher has same behavior

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

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[github-actions]

@Radiancebobo Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[shibd]

trackDelayedDelivery() runs when the dispatcher first reads the entry for delivery, not when the broker stores it. Recomputing deliverAtTime as System.currentTimeMillis() + (clientDeliverAt - publishTime) therefore restarts the full original delay at dispatch time.

For example, a deliverAfter(3m) message that already sat in backlog for 2 minutes will be delayed for another 3 minutes here, instead of only the remaining 1 minute (or immediately if overdue). That changes delayed-delivery semantics and can over-delay backlog / recovery cases.

[Radiancebobo]

Thanks for the review! You're right — trackDelayedDelivery() runs at dispatch time, so System.currentTimeMillis() + relativeDelay restarts the full delay, ignoring time already spent in backlog.

I've moved the correction to publish time (PersistentTopic.publishMessage(), before persisting to the ledger):

correctedDeliverAt = brokerNow + (deliverAtTime - publishTime)

Since both deliverAtTime and publishTime are set by the client, their difference is the intended delay (clock-skew-free). At publish time brokerNow IS the broker-side publish time, so there's no backlog issue. The corrected value is persisted, and dispatch-time logic uses it as-is — no recalculation needed.

The dispatch-time changes in both dispatchers have been reverted to the original one-liner.

[shibd]

Moving the correction to publish time fixes the backlog/redispatch semantic issue, but I still don't think this is a good tradeoff for the broker.

PersistentTopic.publishMessage() is on a very hot path. For delayed messages this now adds a metadata parse, metadata copy, retained payload slice, full metadata+payload re-serialization, and checksum recomputation before the ledger append. That turns a clock-sync/environment problem into a permanent per-message tax in the broker.

More importantly, the broker is still not using an authoritative time source for the requested delay. It is deriving the delay from two client-provided timestamps and assuming they were produced consistently by the client. In practice, if we have a deployment where client and broker clocks are persistently out of sync, the right fix is to keep OS/NTP/PTP time synchronization healthy rather than rewriting message metadata on the broker hot path.

Given the added cost and complexity, I would be inclined to keep delayed-delivery semantics simple here and treat clock skew as an operational issue unless we have strong performance data showing this overhead is negligible and a stronger justification for compensating in the broker.