Class DefaultTrackingStrategy

java.lang.Object
io.fluxzero.common.tracking.DefaultTrackingStrategy
All Implemented Interfaces:
TrackingStrategy, Closeable, AutoCloseable
public class DefaultTrackingStrategy extends Object implements TrackingStrategy
Streaming strategy that allows multiple clients to concurrently consume a message stream. Messages are routed to clients based on the value of their segment. Each connected client handles a distinct range of segments.

Message segments are determined by the clients that publish the messages (usually based on the consistent hash of some routing key, like the value of a user id).

If a client joins or leaves the cluster the segment range mapped to each client is recalculated so messages may get routed differently than before.

Clients can safely join or leave the cluster at any time. The strategy guarantees that a message is not consumed by more than one client.

  • Field Details

    • DEFAULT_INITIAL_POSITION_LAG

      public static final Duration DEFAULT_INITIAL_POSITION_LAG
      Default look-back applied when a consumer has no stored or client-controlled position.

  • Constructor Details

    • DefaultTrackingStrategy

      public DefaultTrackingStrategy(MessageStore source, PositionStore positionStore)

    • DefaultTrackingStrategy

      public DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, Duration initialPositionLag)
      Creates a tracking strategy with a configurable look-back for consumers without a stored position.
      Parameters:
      source - source message log
      positionStore - consumer position store
      initialPositionLag - duration subtracted from the current time for a new consumer

    • DefaultTrackingStrategy

      public DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, String traceMessageType)
      Creates a strategy whose JFR batch metrics identify the logical message log.
      Parameters:
      traceMessageType - logical message type used only while batch recording is enabled

    • DefaultTrackingStrategy

      public DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, TaskScheduler scheduler)

    • DefaultTrackingStrategy

      protected DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, TaskScheduler scheduler, Duration initialPositionLag)

    • DefaultTrackingStrategy

      protected DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, TaskScheduler scheduler, int segments)

    • DefaultTrackingStrategy

      protected DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, TaskScheduler scheduler, int segments, String traceMessageType)

    • DefaultTrackingStrategy

      protected DefaultTrackingStrategy(MessageStore source, PositionStore positionStore, TaskScheduler scheduler, int segments, Duration initialPositionLag)

  • Method Details

    • freezeForHandover

      public List<TrackerClaim> freezeForHandover()
      Freezes ownership before an elected Runtime relinquishes its lease. Concurrent admitted reads finish registering before the snapshot; later reads remain retryable. Socket cleanup cannot erase frozen claims. The caller must persist this snapshot before making the successor available.

    • freezeForHandover

      public List<TrackerClaim> freezeForHandover(Duration timeout)
      Freezes ownership, allowing at most timeout for an admitted storage scan to release its lock. Failure leaves the caller responsible for retaining leadership or treating shutdown as a failed handover.

    • restoreClaims

      public void restoreClaims(List<TrackerClaim> claims)
      Restores exact active ranges before any successor read is admitted.

    • getBatch

      public CompletableFuture<MessageBatch> getBatch(Tracker tracker)
      Description copied from interface: TrackingStrategy
      Requests a new batch of messages for the given tracker.

      This method is typically invoked by the Tracker when it is ready to handle more messages. Depending on the strategy, this method may:

      • Fetch messages directly from a MessageStore and complete the result (e.g. for log tailing), or
      • Suspend the result until messages become available
      Specified by:
      getBatch in interface TrackingStrategy
      Parameters:
      tracker - the tracker requesting a batch
      Returns:
      future completed with the requested batch

    • getBatch

      protected void getBatch(Tracker tracker, DefaultTrackingStrategy.TrackerRequest<MessageBatch> request)

    • claimSegment

      public CompletableFuture<ClaimResult> claimSegment(Tracker tracker)
      Description copied from interface: TrackingStrategy
      Claims one or more message segments for the given tracker.

      This method is invoked when segment-based partitioning is enabled. It ensures that each segment is only claimed by a single tracker at a time and may release conflicting claims if necessary.

      Specified by:
      claimSegment in interface TrackingStrategy
      Parameters:
      tracker - the tracker attempting to claim a segment
      Returns:
      future completed with the claimed segment and current position

    • claimSegment

      protected void claimSegment(Tracker tracker, DefaultTrackingStrategy.TrackerRequest<ClaimResult> request)

    • getBatch

      protected List<SerializedMessage> getBatch(int[] segment, Position position, int batchSize)

    • scanBatch

      protected MessageStoreBatch scanBatch(int[] segment, Position position, int batchSize, long maxBytes, Predicate<? super SerializedMessage> filter)

    • scanBatch

      protected MessageStoreBatch scanBatch(int[] segment, Position position, int batchSize, long maxBytes, Predicate<? super SerializedMessage> filter, boolean includeDocumentTombstones)

    • waitForMessages

      protected void waitForMessages(Tracker tracker, MessageBatch emptyBatch)

    • waitForMessages

      protected void waitForMessages(Tracker tracker, MessageBatch emptyBatch, DefaultTrackingStrategy.TrackerRequest<MessageBatch> request)

    • waitForUpdate

      protected void waitForUpdate(Tracker tracker, MessageBatch emptyBatch, Runnable followUp)

    • waitForUpdate

      protected void waitForUpdate(Tracker tracker, MessageBatch emptyBatch, Runnable followUp, DefaultTrackingStrategy.TrackerRequest<?> request)

    • position

      protected Position position(Tracker tracker, int[] segment)

    • filter

      protected List<SerializedMessage> filter(List<SerializedMessage> messages, int[] segmentRange, Position position, Tracker tracker)

    • filterPredicate

      protected Predicate<SerializedMessage> filterPredicate(int[] segmentRange, Position position, Tracker tracker)

    • ensureMessageSegment

      protected SerializedMessage ensureMessageSegment(SerializedMessage message)

    • adjustMaxSize

      protected int adjustMaxSize(Tracker tracker, int maxSize)

    • claimSegmentRange

      protected int[] claimSegmentRange(Tracker tracker)

    • onUpdate

      protected void onUpdate(List<SerializedMessage> messages)

    • onClusterUpdate

      protected void onClusterUpdate(TrackerCluster cluster)

    • disconnectTrackers

      public Set<Tracker> disconnectTrackers(Predicate<Tracker> predicate, boolean sendFinalEmptyBatch)
      Description copied from interface: TrackingStrategy
      Disconnects trackers that match the provided filter.

      This is typically used during client shutdown, reconfiguration, or error handling to forcibly remove trackers from the strategy's internal registry.

      Specified by:
      disconnectTrackers in interface TrackingStrategy
      Parameters:
      predicate - filter for matching trackers to disconnect
      sendFinalEmptyBatch - if true, a final empty batch should be sent to each disconnected tracker to allow graceful termination
      Returns:
      the trackers that were disconnected (empty if no trackers matched the filter)

    • disconnectTrackersOnClose

      public Set<Tracker> disconnectTrackersOnClose(Predicate<Tracker> predicate)
      Removes socket-owned trackers during connection cleanup. A frozen handover retains their claims; explicit client disconnects must use disconnectTrackers(Predicate, boolean) and remain retryable.

    • purgeCeasedTrackers

      protected void purgeCeasedTrackers(Duration delay)

    • close

      public void close()
      Description copied from interface: TrackingStrategy
      Closes the tracking strategy and releases any underlying resources.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Specified by:
      close in interface TrackingStrategy