Write your stream

Requirements to support Stream Level (High Level) consumers

The stream should provide the following guarantees:

• Exactly once delivery (unless restarting from a checkpoint) for each consumer of the stream.

• (Optionally) support mechanism to split events (in some arbitrary fashion) so that each event in the stream is delivered exactly to one host out of set of hosts.

• Provide ways to save a checkpoint for the data consumed so far. If the stream is partitioned, then this checkpoint is a vector of checkpoints for events consumed from individual partitions.

• The checkpoints should be recorded only when Pinot makes a call to do so.

• The consumer should be able to start consumption from one of:

  • latest avaialble data

  • earliest available data

Requirements to support Partition Level (Low Level) consumers

While consuming rows at a partition level, the stream should support the following properties:

• Stream should provide a mechanism to get the current number of partitions.

• Each event in a partition should have a unique offset that is not more than 64 bits long.

• Refer to a partition as a number not exceeding 32 bits long.

In addition, we have an operational requirement that the number of partitions should not be reduced over time.

Stream plug-in implementation

In order to add a new type of stream (say, Foo) implement the following classes:

1. FooConsumerFactory extends

2. FooPartitionLevelConsumer implements

3. FooStreamLevelConsumer implements

Depending on stream level or partition level, your implementation needs to include StreamLevelConsumer or PartitionLevelConsumer.

The properties for the stream implementation are to be set in the table configuration, inside section.

Use the streamType property to define the stream type. For example, for the implementation of stream foo, set the property "streamType" : "foo".

The rest of the configuration properties for your stream should be set with the prefix "stream.foo". Be sure to use the same suffix for: (see examples below):

• topic

• consumer type

• stream consumer factory

All values should be strings. For example:

You can have additional properties that are specific to your stream. For example:

In addition to these properties, you can define thresholds for the consuming segments:

• rows threshold

• time threshold

The properties for the thresholds are as follows:

An example of this implementation can be found in the , which is an implementation for the kafka stream.