org.apache.samza.system.chooser

Interface MessageChooser

All Known Implementing Classes:

BaseMessageChooser

public interface MessageChooser

MessageChooser is an interface for programmatic fine-grain control over stream consumption.

Consider the case of a Samza task consuming multiple streams, where some streams may be from live systems that have stricter SLA requirements and must always be prioritized over other streams that may be from batch systems. MessageChooser allows developers to inject message prioritization logic into the SamzaContainer.

In general, the MessageChooser can be used to prioritize certain systems, streams or partitions over others. It can also be used to throttle certain partitions, by choosing not to return messages even though they are available. The MessageChooser can also throttle the entire SamzaContainer by performing a blocking operation, such as Thread.sleep.

The manner in which MessageChooser is used is:

• SystemConsumers buffer messages from all SystemStreamPartitions as they become available.
• If MessageChooser has no messages for a given SystemStreamPartition, and a SystemConsumer has a message in its buffer for the SystemStreamPartition, the MessageChooser will be updated once with the next message in the buffer.
• When SamzaContainer is ready to process another message, it calls SystemConsumers.choose, which in turn calls choose().

Since the MessageChooser only receives one message at a time per SystemStreamPartition, it can be used to order messages between different SystemStreamPartitions, but it can't be used to re-order messages within a single SystemStreamPartition (a buffered sort). This must be done within a StreamTask.

The contract between the MessageChooser and the SystemConsumers is:

• update(IncomingMessageEnvelope) can be called multiple times before choose() is called.
• If choose() returns null, that means no envelopes should be processed at the moment.
• A MessageChooser may elect to return null when choose() is called, even if unprocessed messages have been given by the update method.
• A MessageChooser will not have any of its in-memory state restored in the event of a failure.
• Blocking operations (such as Thread.sleep) will block all processing in the entire SamzaContainer.
• A MessageChooser should never return the same envelope more than once.
• Non-deterministic (e.g. time-based) MessageChoosers are allowed.
• A MessageChooser does not need to be thread-safe.

Method Summary

Modifier and Type	Method and Description
IncomingMessageEnvelope	choose() The choose method is invoked when the SamzaContainer is ready to process a new message.
void	register(SystemStreamPartition systemStreamPartition, java.lang.String offset) Called before start, to let the chooser know that it will be handling envelopes from the given SystemStreamPartition.
void	start() Called after all SystemStreamPartitions have been registered.
void	stop() Called when the chooser is about to be discarded.
void	update(IncomingMessageEnvelope envelope) Notify the chooser that a new envelope is available for a processing.

Method Detail

start

void start()

Called after all SystemStreamPartitions have been registered. Start is used to notify the chooser that it will start receiving update and choose calls.

stop

void stop()

Called when the chooser is about to be discarded. No more messages will be given to the chooser after it is stopped.

register

void register(SystemStreamPartition systemStreamPartition,
              java.lang.String offset)

Called before start, to let the chooser know that it will be handling envelopes from the given SystemStreamPartition. Register will only be called before start.

Parameters:

systemStreamPartition - A SystemStreamPartition that envelopes will be coming from.

offset - The offset of the first message expected for the system/stream/partition that's being registered. If "7" were supplied as the offset, then the MessageChooser can expect the first message it is updated with for the system/stream/partition will have an offset of "7".

update

void update(IncomingMessageEnvelope envelope)

Notify the chooser that a new envelope is available for a processing. A MessageChooser will receive, at most, one outstanding envelope per system/stream/partition combination. For example, if update is called for partition 7 of kafka.mystream, then update will not be called with an envelope from partition 7 of kafka.mystream until the previous envelope has been returned via the choose method. Update will only be invoked after the chooser has been started.

Parameters:

envelope - An unprocessed envelope.

choose

IncomingMessageEnvelope choose()

The choose method is invoked when the SamzaContainer is ready to process a new message. The chooser may elect to return any envelope that it's been given via the update method, which hasn't yet been returned. Choose will only be called after the chooser has been started.

Returns:

The next envelope to process, or null if the chooser has no messages or doesn't want to process any at the moment.