Struct KafkaSource 

pub struct KafkaSource { /* private fields */ }

Available on crate features source-kafka and source-rest only.

Implementations

impl KafkaSource

pub async fn new(config: KafkaSourceConfig) -> Result<KafkaSource, FaucetError>

Trait Implementations

impl Source for KafkaSource

fn stream_pages<'a>( &'a self, _context: &'a HashMap<String, Value>, _batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>>

Stream Kafka messages page-by-page. Mirrors the fetch_with_context_incremental poll loop but emits a StreamPage each time the in-memory buffer reaches KafkaSourceConfig::batch_size (or whenever the idle window flushes a partially-filled buffer), and continues polling until the configured max_messages / idle_timeout termination conditions are hit.

The trait-level batch_size argument is intentionally ignored in favour of the config field — the config is the user-facing knob the README documents and routing the pipeline-supplied hint through it would silently override an explicit config value.

Per-page bookmark: each yielded page carries a snapshot of the cumulative (topic, partition) -> next_offset map seen so far. The streaming pipeline persists this via the configured StateStore after the sink confirms the write, giving at-least-once delivery with per-page durability — a crash between pages re-reads only the uncommitted page on resume (the rebalance callback seeds the assigned partitions with the bookmarked offsets before any fetch happens).

batch_size = 0: drains the entire run window (until max_messages or idle_timeout fires) into a single page. This negates the streaming benefit; prefer a finite batch_size in production so the state store advances with each successful sink write.

fn supports_exactly_once(&self) -> bool

Kafka partitions are immutable, ordered logs and every emitted page carries a complete per-partition next-offset bookmark, so resuming from any bookmark continues the record stream at exactly that position — no record before it is re-emitted, none after it is skipped. Page boundaries on replay may differ (idle-timeout cuts are timing- dependent), which is why the pipeline’s atomic-watermark mechanism re-anchors from the bookmark embedded in the sink’s committed token rather than skipping replayed pages by count. In plain (non-member) mode the source never commits consumer-group offsets, so the broker’s group position can never run ahead of the durable bookmark.

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, KafkaSource: 'async_trait,

Preflight probe that does not consume any message.

The default Source::check would call stream_pages, which polls for a message and would block on an empty topic until the idle/max-message terminator fires. Instead we fetch cluster metadata (fetch_metadata(None, timeout)), which validates broker connectivity + auth without consuming or committing anything.

fetch_metadata is a blocking librdkafka call, so it runs on a blocking thread; the whole probe is additionally bounded by ctx.timeout.

fn is_shardable(&self) -> bool

The Kafka source is always shardable via native consumer-group assignment (#261): each shard is a membership slot in the shared group, and the broker — not faucet — assigns partitions across the members and rebalances on membership change. Sharding only takes effect when the cluster coordinator calls apply_shard; a plain faucet run consumes as a single group member, unchanged.

fn enumerate_shards<'life0, 'async_trait>( &'life0 self, target: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ShardSpec>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, KafkaSource: 'async_trait,

Enumerate target group-membership slots, capped at the subscription’s total partition count (extra members would never be assigned a partition). Falls back to the uncapped target when topic metadata cannot be resolved.

fn apply_shard<'life0, 'life1, 'async_trait>( &'life0 self, shard: &'life1 ShardSpec, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, KafkaSource: 'async_trait,

Enter group-member mode for one membership slot (or leave it, for the whole-dataset shard). In member mode the source additionally:

• commits offsets to the consumer group at durable page boundaries (see stream_pages), so partitions that migrate between members resume from the last durably-written position;
• defers bookmark seeks to the group’s committed offsets whenever those are ahead (another member may have advanced a partition past this member’s bookmark).

Only the streaming path (stream_pages, which every Pipeline run drives) commits; the batch fetch_all path never commits because its records are not durable until after the fetch returns.

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, context: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, KafkaSource: 'async_trait,

Primary fetch method. Receives context from a parent source’s records.

fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, _context: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, KafkaSource: 'async_trait,

Incremental fetch with parent context support.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

fn state_key(&self) -> Option<String>

Stable key under which this source’s incremental-replication bookmark should be persisted in a StateStore.

fn apply_start_bookmark<'life0, 'async_trait>( &'life0 self, bookmark: Value, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, KafkaSource: 'async_trait,

Apply a bookmark loaded from a StateStore as this run’s starting point.

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. Defaults to the final segment of std::any::type_name::<Self>(), e.g. "RestSource". Built-in connectors override with a short, friendly snake_case name (e.g. "rest"). Must return a non-empty string; observability decorators fall back to "unknown" in release builds if it is empty (and debug_assert! in debug builds).

fn dataset_uri(&self) -> String

Logical dataset identity for lineage emission, following OpenLineage naming conventions (https://openlineage.io/docs/spec/naming).

fn fetch_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: fetch with no parent context.

fn fetch_all_incremental<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: incremental fetch with no parent context.

fn capture_resume_position<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Capture the source’s current replication position without consuming any changes, ensuring any server-side resource (e.g. a logical replication slot) needed to later resume from that position exists.

fn replay_guarantee(&self) -> ReplayGuarantee

The typed replay capability this source advertises — see ReplayGuarantee.

fn supports_discover(&self) -> bool

Whether this source can enumerate the datasets behind its connection via discover. Default: false. Sources backed by an introspectable catalog (database information_schema, MongoDB collections, Elasticsearch indices, object-store prefixes) override this to true.

fn discover<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<DatasetDescriptor>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Enumerate the datasets living behind this source’s connection — one DatasetDescriptor per table / collection / index / prefix, each carrying a partial config override that selects it (used by faucet discover to scaffold one matrix row per dataset).