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.

  • Constructor Details

  • Method Details

    • getBatch

      public void getBatch(Tracker tracker, PositionStore positionStore)
      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 deliver them to the tracker (e.g. for log tailing), or
      • Suspend the tracker until messages become available
      Specified by:
      getBatch in interface TrackingStrategy
      Parameters:
      tracker - the tracker requesting a batch
      positionStore - to fetch or update tracking positions

    • claimSegment

      public void claimSegment(Tracker tracker, PositionStore positionStore)
      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
      positionStore - to fetch tracking positions

    • getBatch

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

    • waitForMessages

      protected void waitForMessages(Tracker tracker, MessageBatch emptyBatch, PositionStore positionStore)

    • waitForUpdate

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

    • position

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

    • filter

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

    • ensureMessageSegment

      protected SerializedMessage ensureMessageSegment(SerializedMessage message)

    • adjustMaxSize

      protected int adjustMaxSize(Tracker tracker, int maxSize)

    • claimSegment

      protected int[] claimSegment(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)

    • 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