Struct KafkaAdapter 

pub struct KafkaAdapter { /* private fields */ }

Kafka transport adapter for event streaming.

Delivers subscription events to Apache Kafka topics. Uses the entity_id as the message key for consistent partitioning.

Feature Flag

This adapter has two implementations:

• With kafka feature: delivery through fraiseql_kafka::KafkaEgress, the one producer this workspace builds
• Without kafka feature: a stub that fails loud on every delivery (#784)

Example

use fraiseql_core::runtime::subscription::{KafkaAdapter, KafkaConfig};

let config = KafkaConfig::new("kafka+ssl://localhost:9093", "fraiseql-events")
    .with_client_id("my-service")
    .with_compression("lz4");

let adapter = KafkaAdapter::new(config)?;
adapter.deliver(&event, "orderCreated").await?;

Implementations

impl KafkaAdapter

pub fn new(config: KafkaConfig) -> Result<Self, SubscriptionError>

Connect this adapter’s shared Kafka egress.

The endpoint is screened and security.protocol set from its scheme before any client exists — by fraiseql_kafka::KafkaEgress, the one place in this workspace that builds a producer, so this transport and the CDC outbox sink cannot disagree about what is safe (#1102).

Errors

SubscriptionError::Internal if the endpoint is refused (no scheme, unsupported scheme, unopted-in plaintext, blocked broker host, unresolvable SASL credentials) or if the producer cannot be created.

Trait Implementations

impl Debug for KafkaAdapter

Available on crate feature kafka only.

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl TransportAdapter for KafkaAdapter

fn deliver<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, event: &'life1 SubscriptionEvent, subscription_name: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), SubscriptionError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Deliver an event to the transport.

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

Get the adapter name for logging/metrics.

fn health_check<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = bool> + Send + 'async_trait>>

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

Check if the adapter is healthy/connected.