Stream Ingestion Plugin

This page describes how to write your own stream ingestion plugin for Pinot.

You can write custom stream ingestion plugins to add support for your own streaming platforms such as Pravega, Kinesis etc.

Pinot consumes each stream partition independently, tracking offsets per partition. A stream you want to integrate must therefore expose per-partition offsets and support consuming from a given offset.

Requirements to support a stream

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.

• Stream should provide the following mechanisms to get an offset for a given partition of the stream:

  • get the offset of the oldest event available (assuming events are aged out periodically) in the partition.

  • get the offset of the most recent event published in the partition

  • (optionally) get the offset of an event that was published at a specified time

• Stream should provide a mechanism to consume a set of events from a partition starting from a specified offset.

• Pinot assumes that the offsets of incoming events are monotonically increasing; i.e., if Pinot consumes an event at offset o1, then the offset o2 of the following event should be such that o2 > o1.

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 StreamConsumerFactory

2. FooPartitionGroupConsumer implements PartitionGroupConsumer

3. FooMetadataProvider implements StreamMetadataProvider

4. FooMessageDecoder implements StreamMessageDecoder

The properties for the stream implementation are to be set in the table configuration, inside streamConfigs 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

• stream consumer factory

• offset

• decoder class name

• decoder properties

• connection timeout

• fetch timeout

All values should be strings. For example:

"streamType" : "foo",
"stream.foo.topic.name" : "SomeTopic",
"stream.foo.consumer.factory.class.name": "fully.qualified.pkg.ConsumerFactoryClassName",
"stream.foo.consumer.prop.auto.offset.reset": "largest",
"stream.foo.decoder.class.name" : "fully.qualified.pkg.DecoderClassName",
"stream.foo.decoder.prop.a.decoder.property" : "decoderPropValue",
"stream.foo.connection.timeout.millis" : "10000", // default 30_000
"stream.foo.fetch.timeout.millis" : "10000" // default 5_000

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

"stream.foo.some.buffer.size" : "24g"

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:

"realtime.segment.flush.threshold.rows" : "100000"
"realtime.segment.flush.threshold.time" : "6h"

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