kafkit-client 0.1.9

Kafka 4.0+ pure Rust client.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

# Security

`kafkit-client` supports TLS and SASL on producer, consumer, share-consumer, and
admin connections.

Supported mechanisms:

- TLS with system roots.
- TLS with custom CA files.
- TLS server-name override.
- TLS client certificates.
- SASL/PLAIN.
- SASL/SCRAM-SHA-256.
- SASL/SCRAM-SHA-512.

Unsupported mechanisms:

- OAuthBearer.
- GSSAPI/Kerberos.
- AWS/MSK IAM.

These unsupported mechanisms are intentionally out of scope for the initial
release.

## TLS

Use TLS only against Kafka listeners configured for TLS.

```rust,no_run
use kafkit_client::{KafkaClient, TlsConfig};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let producer = KafkaClient::new("kafka.example.com:9092")
        .topic("orders")
        .producer()
        .with_tls(
            TlsConfig::new()
                .with_ca_cert_path("/etc/ssl/certs/kafka-ca.pem")
                .with_server_name("kafka.example.com"),
        )
        .connect()
        .await?;

    producer.shutdown().await?;
    Ok(())
}
```

For mutual TLS, add client certificate and key paths:

```rust,no_run
use kafkit_client::{KafkaAdmin, AdminConfig};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let admin = KafkaAdmin::connect(
        AdminConfig::new("kafka.example.com:9093")
            .with_tls_ca_cert_path("/etc/ssl/certs/kafka-ca.pem")
            .with_tls_client_auth_paths("/etc/kafka/client.crt", "/etc/kafka/client.key")
            .with_tls_server_name("kafka.example.com"),
    )
    .await?;

    let cluster = admin.describe_cluster().await?;
    println!("cluster id: {}", cluster.cluster_id);
    Ok(())
}
```

## SASL

SASL can be used by itself or combined with TLS. Most production deployments
should combine SASL with TLS so credentials are protected in transit.

```rust,no_run
use kafkit_client::KafkaClient;

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let producer = KafkaClient::new("kafka.example.com:9093")
        .topic("orders")
        .producer()
        .with_sasl_scram_sha_512("alice", "correct-horse-battery-staple")
        .connect()
        .await?;

    producer.shutdown().await?;
    Ok(())
}
```

Available helpers:

- `with_sasl_plain(username, password)`
- `with_sasl_scram_sha_256(username, password)`
- `with_sasl_scram_sha_512(username, password)`

The broker-backed integration suite covers TLS with custom CAs, server-name
override, mutual TLS client certificates, SASL/PLAIN, SASL/SCRAM-SHA-256, and
SASL/SCRAM-SHA-512. It also checks representative authentication failures do
not include passwords or TLS store secrets in returned error messages.

## Troubleshooting Security

- Verify the bootstrap address points at the listener type you configured.
- Use TLS settings only for TLS listeners.
- Use SASL settings only for SASL-enabled listeners.
- Check the server name against the broker certificate subject/SAN.
- Check CA files and client cert/key paths are readable by the process.