Struct dataflow::source::KinesisSourceReader

pub struct KinesisSourceReader {
    kinesis_client: KinesisClient,
    stream_name: String,
    shard_set: HashMap<String, ShardMetrics>,
    shard_queue: VecDeque<(String, Option<String>)>,
    last_checked_shards: Instant,
    buffered_messages: VecDeque<SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>>,
    processed_message_count: i64,
    base_metrics: KinesisMetrics,
}

Contains all information necessary to ingest data from Kinesis

Fields

kinesis_client: KinesisClient

Kinesis client used to obtain records

stream_name: String

The name of the stream

shard_set: HashMap<String, ShardMetrics>

The set of active shards

shard_queue: VecDeque<(String, Option<String>)>

A queue representing the next shard to read from. This is necessary to ensure that all shards are read from uniformly

last_checked_shards: Instant

The time at which we last refreshed metadata TODO(natacha): this should be moved to timestamper

buffered_messages: VecDeque<SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>>

Storage for messages that have not yet been timestamped

processed_message_count: i64

Count of processed message

base_metrics: KinesisMetrics

Metrics from which per-shard metrics get created.

Implementations

impl KinesisSourceReader

async fn update_shard_information(&mut self) -> Result<(), Error>

async fn get_records(
    &self,
    shard_iterator: &str
) -> Result<GetRecordsOutput, SdkError<GetRecordsError>>

Obtains the next record for this shard given a shard iterator

Trait Implementations

impl SourceReader for KinesisSourceReader

type Key = Option<Vec<u8>>

type Value = Option<Vec<u8>>

fn new(
    _source_name: String,
    _source_id: SourceInstanceId,
    _worker_id: usize,
    _worker_count: usize,
    _consumer_activator: SyncActivator,
    connector: ExternalSourceConnector,
    _restored_offsets: Vec<(PartitionId, Option<MzOffset>)>,
    _encoding: SourceDataEncoding,
    _: Option<Logger>,
    base_metrics: SourceBaseMetrics
) -> Result<(Self, Option<PartitionId>), Error>

Create a new source reader.

fn get_next_message(
    &mut self
) -> Result<NextMessage<Self::Key, Self::Value>, Error>

Returns the next message available from the source.