kafka_client

use kafka_client::Client;

let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .with_client_id("my-app")
    .build()
    .await?;

// Query cluster metadata
let brokers = client.metadata().get_all_brokers().await;
println!("Discovered {} brokers", brokers.len());

use kafka_client::{Client, ProducerConfig, ProducerRecord};
use bytes::Bytes;

let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .build()
    .await?;

// Default config
let producer = client.producer_default().await;

// Or with custom config
let producer = client.producer(
    ProducerConfig::new().with_acks(-1).with_retries(3)
).await;

// Send messages
let record = ProducerRecord::new("my-topic", Bytes::from("hello world"));
let metadata = producer.send(record).await?;
println!("Sent to partition {} at offset {}", metadata.partition, metadata.offset);

// Ensure all messages are delivered
producer.flush().await?;

use kafka_client::{Client, ConsumerConfig};

let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .build()
    .await?;

// Default group consumer
let mut consumer = client.consumer_default();

// Or with custom config
let mut consumer = client.consumer(
    ConsumerConfig::new("my-consumer-group")
        .with_earliest()
);

consumer.subscribe(vec!["my-topic".to_string()]).await?;

// Poll for messages
let records = consumer.poll_timeout(Duration::from_millis(5000)).await?;
for record in records {
    println!("Received: {:?}", record.value);
}

use kafka_client::{Client, admin::NewTopic};

let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .build()
    .await?;
let admin = client.admin();

// Create a topic
admin.create_topic(&NewTopic::new("orders", 3, 3)).await?;

// List all topics
let topics = admin.list_topics().await?;
for t in &topics { println!("{}", t.name); }

// Describe cluster
let info = admin.describe_cluster().await?;
println!("{} brokers, controller: {:?}", info.brokers.len(), info.controller_id);

use kafka_client::{Client, SaslMechanismType};

// PLAIN authentication
let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .with_sasl(SaslMechanismType::Plain, "username", "password")
    .build()
    .await?;

// SCRAM-SHA-256 authentication
let client = Client::builder(vec!["127.0.0.1:9092".parse()?])
    .with_sasl(SaslMechanismType::ScramSha256, "username", "password")
    .build()
    .await?;

use kafka_client::{Client, KerberosCredentials};

let client = Client::builder(vec!["127.0.0.1:9096".parse()?])
    .with_kerberos(
        KerberosCredentials::new("client@EXAMPLE.COM")
            .with_keytab("/path/to/client.keytab")
    )
    .with_kdc("kdc.example.com", 88)
    .with_broker_hostname("kafka-broker.example.com")
    .build()
    .await?;

use kafka_client::{Client, TlsConfig};

let tls = TlsConfig {
    domain: "kafka.example.com".to_string(),
    ..Default::default()
};

// TLS only
let client = Client::builder(vec!["127.0.0.1:9093".parse()?])
    .with_tls_config(tls)
    .build()
    .await?;

// TLS + SASL
let client = Client::builder(vec!["127.0.0.1:9093".parse()?])
    .with_sasl_tls(tls, SaslMechanismType::ScramSha256, "username", "password")
    .build()
    .await?;