连接生产集群

生产集群要求认证和 TLS，服务则要设置生产者的持久性和自己的超时。

基于 TLS 的 SASL

src/main.rs

/// Where the cluster is and who the service is, as the deployment provides them.
struct Cluster {
    servers: String,
    username: String,
    password: String,
}

impl Cluster {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            servers: read("KAFKA_SERVERS"),
            username: read("KAFKA_USERNAME"),
            password: read("KAFKA_PASSWORD"),
        }
    }
}

fn broker(cluster: Cluster) -> KafkaBroker {
    KafkaBroker::new([cluster.servers])
        .default_group("orders-svc")
        .config("security.protocol", "SASL_SSL")
        .config("sasl.mechanism", "SCRAM-SHA-512")
        .config("sasl.username", cluster.username)
        .config("sasl.password", cluster.password)
        .producer_config("acks", "all")
        .producer_config("enable.idempotence", "true")
        .connect_timeout(Duration::from_secs(10))
        .flush_timeout(Duration::from_secs(20))
}

Cargo.toml

[dependencies]
ruststream-rdkafka = { version = "0.7", features = ["ssl"] }

部署环境通过环境变量提供地址和凭据。 每个 librdkafka 属性都经由 KafkaBroker::config 设置，它把属性应用到各个消费者和生产者上。 security.protocol 选择基于 TLS 的 SASL，sasl.mechanism 选择 SASL 机制。 TLS 要求 librdkafka 以 OpenSSL 构建：ssl feature 链接系统的 OpenSSL，ssl-vendored 从源码构建一个。 SCRAM 和 OAUTHBEARER 需要同样的 feature，因为 librdkafka 基于 OpenSSL 实现它们。 基于 SASL_PLAINTEXT 的 SASL PLAIN 不需要任何 feature。 Kerberos（GSSAPI）是 rust-rdkafka 的 gssapi feature：把 rdkafka 连同该 feature 加入你的依赖。 使用 TLS 而不使用 SASL 的集群，把 security.protocol 设为 SSL，由 ssl.* 属性指定证书。

持久性与超时

producer_config 只在生产者上设置属性。 acks=all 让集群在每个同步副本都拿到记录后才确认它。 enable.idempotence 防止生产者重试把一条记录写入两次。 每次发布都等待集群的确认，所以发布成功意味着集群已经接受了记录。

Broker 有自己的超时：

• connect_timeout 限制启动时检查集群是否可达的时间。
• flush_timeout 限制关闭时刷出未发送记录的时间。
• assignment_timeout 限制启动时等待组为订阅分配分区的时间。

服务

宏

src/main.rs

#[subscriber(KafkaTopic::new("orders").commit(Commit::Tracked), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { id: order.id }
}

fn service(broker: KafkaBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm)
            .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker(Cluster::from_env()))
}

手写

src/main.rs

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn service(broker: KafkaBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("orders", Confirm)
                .map_source(|_| KafkaTopic::new("orders").commit(Commit::Tracked))
                .reply()
                .build(),
        )
        .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
    })
}

fn app() -> RustStream {
    service(broker(Cluster::from_env()))
}

fn main() -> ExitCode {
    run_main(app)
}

librdkafka 的本地队列已满时，发布会等待队列腾出空间。 发布策略上的 queue_timeout 把更长的等待变成错误。

订阅用 KafkaTopic::config 设置自己的属性，它们优先于 Broker 的属性。 有两个属性属于提交模式：enable.auto.commit 和 enable.auto.offset.store。 在 Commit::Tracked 和 Commit::Transactional 下，设置了其中任何一个的订阅都拒绝启动，错误信息会指出该属性和模式。

测试配置

测试在 TestApp::start 下启动生产环境的 service，配上生产环境的 broker 和测试凭据。 进程内 Kafka 像 librdkafka 那样校验配置，但不会去连接集群。 librdkafka 拒绝的属性名或属性值会让启动失败，并带上 librdkafka 的错误消息：

Invalid value "SASL_TLS" for configuration property "security.protocol"

测试需要与服务相同的 feature。 不带 OpenSSL 构建的 librdkafka 会拒绝把 SASL_SSL 和 SSL 作为 security.protocol 的值。 认证本身只在真实集群上进行。