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.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	DefaultTrackingStrategy.TrackerRequest<T>
protected class	DefaultTrackingStrategy.WaitingTracker

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method	Description
protected int	adjustMaxSize(Tracker tracker, int maxSize)
CompletableFuture<ClaimResult>	claimSegment(Tracker tracker)	Claims one or more message segments for the given tracker.
protected void	claimSegment(Tracker tracker, DefaultTrackingStrategy.TrackerRequest<ClaimResult> request)
protected int[]	claimSegmentRange(Tracker tracker)
void	close()	Closes the tracking strategy and releases any underlying resources.
Set<Tracker>	disconnectTrackers(Predicate<Tracker> predicate, boolean sendFinalEmptyBatch)	Disconnects trackers that match the provided filter.
protected SerializedMessage	ensureMessageSegment(SerializedMessage message)
protected List<SerializedMessage>	filter(List<SerializedMessage> messages, int[] segmentRange, Position position, Tracker tracker)
protected Predicate<SerializedMessage>	filterPredicate(int[] segmentRange, Position position, Tracker tracker)
protected List<SerializedMessage>	getBatch(int[] segment, Position position, int batchSize)
CompletableFuture<MessageBatch>	getBatch(Tracker tracker)	Requests a new batch of messages for the given tracker.
protected void	getBatch(Tracker tracker, DefaultTrackingStrategy.TrackerRequest<MessageBatch> request)
protected void	onClusterUpdate(TrackerCluster cluster)
protected void	onUpdate(List<SerializedMessage> messages)
protected Position	position(Tracker tracker, int[] segment)
protected void	purgeCeasedTrackers(Duration delay)
protected MessageStoreBatch	scanBatch(int[] segment, Position position, int batchSize, long maxBytes, Predicate<? super SerializedMessage> filter)
protected void	waitForMessages(Tracker tracker, MessageBatch emptyBatch)
protected void	waitForMessages(Tracker tracker, MessageBatch emptyBatch, DefaultTrackingStrategy.TrackerRequest<MessageBatch> request)
protected void	waitForUpdate(Tracker tracker, MessageBatch emptyBatch, Runnable followUp)
protected void	waitForUpdate(Tracker tracker, MessageBatch emptyBatch, Runnable followUp, DefaultTrackingStrategy.TrackerRequest<?> request)

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

DefaultTrackingStrategy

public DefaultTrackingStrategy(MessageStore source,
 PositionStore positionStore)

DefaultTrackingStrategy

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

DefaultTrackingStrategy

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

Method Details

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)

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)

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