Page History

Image Added

How it works:

1. The upstream tasks will send out control message to all the downstream intermediate topic partition. The control message will be serialized and sent out with user messages in the same stream.

2. Downstream Samza processor will consume the intermediate streams, and deserialize both user messages and control messages in SystemConsumers.

3. The control messages will be reconciled based on the count from all the producers (tasks) from the upstream. See below for more details of different control message reconsiliation. 

Image Removed

How it works for end-of-stream:

1. When an input stream is consumed to the end, Samza send an Eos watermark message to the control channel which includes the input topic and partition.

2. Once all the EOS messages are received (based on the input partition count), we know the input is end-of-stream. For any following intermediate stream whose input streams are all end-of-stream, it will be marked as pending EOS. After that, whenever a marked intermediate stream partition reaches its highest offset (high watermark in Kafka), we can emit end-of-stream message for this partition. It’s guaranteed that the partition is end of stream.

Approach 2: In-band watermark messages

In this approach we don’t use a separate stream to keep the watermark messages. We use the intermediate streams themselves as both data and watermark.

Image Removed

How it works for end-of-stream:

1. When an input stream is consumed to the end, Samza finds out the following intermediate streams that all the inputs have been end-of-stream (through the topology of the operator graph).

2. The task will send an Eos watermark message to all the partitions of the intermediate streams in 1.

3. Each consumer of the intermediate streams will count the watermark messages received for each partition and declare end of stream once all the EOS messages have been received.

Comparisons of the two approaches:

Pros

Cons

Approach 1

- Intermediate streams are clean with only user data. This is convenient if user wants to consume it elsewhere.

- Simple recovery from failure, just read the control stream from the beginning.

- Less number of messages.

- Need to correlate the out-of-band control message with the source stream, which is complex to track and requires synchronization between input streams and control stream. 

- Need to maintain a separate stream for control messages

Approach 2

- No synchronization needed between watermark message and input messages. The watermark can conclude the input messages before this watermark have been complete.This is critical to support general event-time watermarks.

- Complicated failure scenario of the second job. It needs to checkpoint all the watermark messages received, so when it recovered from failure, it can still count.

- More messages required to write to each partition of the downstream processor.

Based on the pros and cons above, we propose to use the in-band approach to support watermarks.

Intermediate Stream Message Format:

The format of the intermediate stream message:

IntermediateMessage =>  [MessageType MessageData]
  MessageType => byte
  MessageData => byte[]

  MessageType => [0(UserMessage), 1(Watermark), 2(EndOfStream)]
  MessageData => [UserMessage/

ControlMessage]
  

ControlMessage => [EndOfStreamMessage/WatermarkMessage]
     

Version => int
     

TaskName => 

String
     

TaskCount => 

int
     Other Message 

Data (based on different types of control message) 

For user message, we will use the user provided serde (default is the system serde). For

control message, we will use JSON serde since it is built in Samza and easy to parse.

Reconciliation

The reconciliation of control messages happens inside TaskInstance after the message is delivered to it from the chooser.  For the scope of this proposal, we support two kinds of control messages: end-of-stream and watermark. 

• End-of-stream Message: This message indicate the upstream task has ended producing to this stream.
• Watermark Message: This message contains a timestamp of the upstream task has processed so far. 

The reconciliation process works as follows:

1. The downstream TaskInstance receives the control message, and update the internal bookkeeping of the messages. For end-of-stream, it keeps the set of upstream tasks for the intermediate stream. For watermark, it keeps the mapping from task to its latest timestamp.

2. Once the task count in the bookkeeping matches the total count, the TaskInstance will emit a single IncomingMessageEnvelope containing the intermediate stream and partition, and the message itself. The timestamp in the watermark message will be: 

   InputWatermark = min { OutputWatermark(task) for each task in upstream tasks }

3. After reconsiliation, the control message evelope will be sent to the task to process.
   

The TaskInstance uses the following maps for bookkeeping received end-of-stream and watermark messages:

EndOfStream Bookkeeping: Map( streamId -> { Set<TaskName>, totalTasks } )
 
Watermark Bookkeeping: Map( streamId -> { Map<TaskName, Timestamp>, totalTasks, timestampOfLastEmission } )

Checkpoint control messages

For failure scenario, we need to keep the state of bookkeeping so we can restore it during recovery. This can be done by checkpointing the bookkeeping states along with the input messages offset.

The checkpoint for EndOfStream:

EndOfStreamCheckpoint =>
 streamId => String
 totalTasks => int
 tasks => Set<String>

The checkpoint for Watermark:

WatermarkCheckpoint =>
 streamId => String
 totalTasks => int
 tasksToEventTime => Map<String, Long>

Detail details

public abstract class ControlMessage {
 

public class WatermarkMessage {
 private final int type;
 private final long timestamp;
 private final String taskName;
  private final int taskCount;

  private WatermarkMessage(int type, long timestamp, version = 1;

  public ControlMessage(String taskName, int taskCount) {
   this.type = type;
   this.timestamp = timestamp;
   this.taskName = taskName;
    this.taskCount = taskCount;
  }

  public intString getTypegetTaskName() {
    return typetaskName;
  }

  public longint getTimestampgetTaskCount() {
    return timestamptaskCount;
  }

  public Stringvoid getTaskNamesetVersion(int version) {
   return taskName; this.version = version;
  }

  public int getTaskCountgetVersion() {
    return taskCountversion;
  }  
}
 
public class EndOfStreamMessage public static WatermarkMessage endOfStream(extends ControlMessage{

 private EndOfStreamMessage(String streamId, String taskName, int taskCount) {
   return new WatermarkMessage(WatermarkType.END_OF_STREAM.ordinal(), Long.MAX_VALUE,   taskName, taskCount)super(taskName, taskCount);
   this.streamId = streamId;
 }
}
 
public staticclass WatermarkMessage eventTimeextends ControlMessage{
 private final long timestamp;

 private WatermarkMessage(long timestamp, String taskName, int taskCount) {
   return new WatermarkMessage(WatermarkType.EVENT_TIME.ordinal(), timestamp, taskName, taskCount);
super(taskName, taskCount);
   this.timestamp = timestamp;
 }

 public long getTimestamp() {   return timestamp;  }
}

Key => IntermediateStreamPartition.WatermarkType
Value => WatermarkCheckpoint
 
public class WatermarkCheckpoint {
 int taskCount;
 Map<String, Long> tasksToEventTime;
}