Подключение к промышленному кластеру

Промышленный кластер требует аутентификации и TLS, а сервис настраивает надёжность своего продюсера и собственные тайм-ауты.

SASL поверх TLS

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 выбирает SASL поверх TLS, а sasl.mechanism - механизм SASL. Для TLS librdkafka должна быть собрана с OpenSSL: фича ssl подключает системный OpenSSL, а ssl-vendored собирает его из исходников. SCRAM и OAUTHBEARER требуют той же фичи, потому что librdkafka строит их на OpenSSL. SASL PLAIN поверх SASL_PLAINTEXT никакой фичи не требует. Kerberos (GSSAPI) - это фича gssapi крейта rust-rdkafka: добавьте rdkafka в свои зависимости с этой фичей. Кластеру с TLS без SASL нужен security.protocol со значением SSL, а сертификаты называют свойства ssl.*.

Надёжность и тайм-ауты

producer_config задаёт свойство только продюсеру. acks=all заставляет кластер подтверждать запись, когда её получили все синхронные реплики. enable.idempotence не даёт повторной отправке продюсера записать запись дважды. Каждая публикация ждёт подтверждения от кластера, поэтому успешная публикация означает, что кластер принял запись.

У брокера есть собственные тайм-ауты:

• 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, и они важнее свойств брокера. Два свойства принадлежат режиму фиксации: enable.auto.commit и enable.auto.offset.store. При Commit::Tracked и Commit::Transactional подписка, которая задаёт любое из них, не запускается, и ошибка называет свойство и режим.

Проверка конфигурации

Тесты запускают рабочий service с рабочим broker, которому переданы тестовые учётные данные, через TestApp::start. Kafka внутри процесса проверяет конфигурацию так же, как librdkafka, но к кластеру не подключается. Имя или значение свойства, которое librdkafka отвергает, проваливает запуск с сообщением librdkafka:

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

Тестам нужны те же фичи, что и сервису. librdkafka, собранная без OpenSSL, отвергает значения SASL_SSL и SSL у security.protocol. Сама аутентификация происходит только на настоящем кластере.