Struct mz_storage::source::kinesis::KinesisSourceReader

pub struct KinesisSourceReader {
    tokio_handle: TokioHandle,
    kinesis_client: KinesisClient,
    stream_name: String,
    shard_set: BTreeMap<String, ShardMetrics>,
    shard_queue: VecDeque<(String, Option<String>)>,
    last_checked_shards: Instant,
    buffered_messages: VecDeque<(SourceMessage<(), Option<Vec<u8>>>, MzOffset)>,
    processed_message_count: u64,
    base_metrics: KinesisMetrics,
    active_read_worker: bool,
    data_capability: Capability<MzOffset>,
    upper_capability: Capability<MzOffset>,
}

Contains all information necessary to ingest data from Kinesis

Fields

tokio_handle: TokioHandle

kinesis_client: KinesisClient

Kinesis client used to obtain records

stream_name: String

The name of the stream

shard_set: BTreeMap<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>>>, MzOffset)>

Storage for messages that have not yet been timestamped

processed_message_count: u64

Count of processed message

base_metrics: KinesisMetrics

Metrics from which per-shard metrics get created.

active_read_worker: bool

data_capability: Capability<MzOffset>

Capabilities used to produce messages

upper_capability: Capability<MzOffset>

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 = ()

type Value = Option<Vec<u8, Global>>

type Time = MzOffset

type Diff = ()

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

Returns the next message available from the source.

fn next<'life0, 'async_trait>(
    &'life0 mut self,
    timestamp_granularity: Duration
) -> Pin<Box<dyn Future<Output = Option<SourceMessageType<Self::Key, Self::Value, Self::Time, Self::Diff>>> + 'async_trait>>where
    Self: 'async_trait,
    'life0: 'async_trait,

Returns the next message available from the source.

fn into_stream<'a>(
    self,
    timestamp_granularity: Duration
) -> LocalBoxStream<'a, SourceMessageType<Self::Key, Self::Value, Self::Time, Self::Diff>>where
    Self: Sized + 'a,

Returns an adapter that treats the source as a stream.