Producer Guide

Producers send messages to Danube topics. This guide covers creating producers, sending messages, partitions, and reliable dispatch.

Creating a Producer

use danube_client::DanubeClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = DanubeClient::builder()
        .service_url("http://127.0.0.1:6650")
        .build()
        .await?;

    let mut producer = client
        .new_producer()
        .with_topic("/default/my-topic")
        .with_name("my-producer")
        .build()?;

    producer.create().await?;

    Ok(())
}

client, err := danube.NewClient().ServiceURL("127.0.0.1:6650").Build()
if err != nil {
    log.Fatalf("Failed to create client: %v", err)
}

ctx := context.Background()

producer, err := client.NewProducer().
    WithName("my-producer").
    WithTopic("/default/my-topic").
    Build()
if err != nil {
    log.Fatalf("Failed to build producer: %v", err)
}

if err := producer.Create(ctx); err != nil {
    log.Fatalf("Failed to create producer: %v", err)
}

DanubeClient client = DanubeClient.builder()
        .serviceUrl("http://127.0.0.1:6650")
        .build();

Producer producer = client.newProducer()
        .withTopic("/default/my-topic")
        .withName("my-producer")
        .build();

producer.create();

client = await (
    DanubeClientBuilder()
    .service_url("http://127.0.0.1:6650")
    .build()
)

producer = (
    client.new_producer()
    .with_topic("/default/my-topic")
    .with_name("my-producer")
    .build()
)

await producer.create()

Topic: destination for messages (e.g., /default/my-topic)
Name: unique producer identifier within the topic
Create: registers the producer with the broker before sending

Sending Messages

Byte Messages

let message_id = producer
    .send("Hello Danube!".as_bytes().to_vec(), None)
    .await?;

println!("Sent message ID: {}", message_id);

messageID, err := producer.Send(ctx, []byte("Hello Danube!"), nil)
if err != nil {
    log.Fatalf("Failed to send: %v", err)
}

fmt.Printf("Sent message ID: %v\n", messageID)

long messageId = producer.send("Hello Danube!".getBytes(), Map.of());

System.out.printf("Sent message ID: %d%n", messageId);

message_id = await producer.send("Hello Danube!".encode())

print(f"Sent message ID: {message_id}")

Messages with Attributes

Attach key-value metadata to messages for routing hints, custom headers, or tracing context:

use std::collections::HashMap;

let mut attributes = HashMap::new();
attributes.insert("source".to_string(), "app-1".to_string());
attributes.insert("priority".to_string(), "high".to_string());

let message_id = producer
    .send(b"Important message".to_vec(), Some(attributes))
    .await?;

attributes := map[string]string{
    "source":   "app-1",
    "priority": "high",
}

messageID, err := producer.Send(ctx, []byte("Important message"), attributes)

Map<String, String> attributes = Map.of(
        "source",   "app-1",
        "priority", "high"
);

long messageId = producer.send("Important message".getBytes(), attributes);

attributes = {
    "source": "app-1",
    "priority": "high",
}

message_id = await producer.send(b"Important message", attributes)

Partitioned Topics

Partitions enable horizontal scaling by distributing messages across multiple brokers. Add .with_partitions(N) to the producer builder:

let mut producer = client
    .new_producer()
    .with_topic("/default/high-throughput")
    .with_name("partitioned-producer")
    .with_partitions(3)
    .build()?;

producer.create().await?;

producer, err := client.NewProducer().
    WithName("partitioned-producer").
    WithTopic("/default/high-throughput").
    WithPartitions(3).
    Build()

producer.Create(ctx)

Producer producer = client.newProducer()
        .withTopic("/default/high-throughput")
        .withName("partitioned-producer")
        .withPartitions(3)
        .build();

producer.create();

producer = (
    client.new_producer()
    .with_topic("/default/high-throughput")
    .with_name("partitioned-producer")
    .with_partitions(3)
    .build()
)

await producer.create()

The topic splits into N internal partitions (e.g., -part-0, -part-1, -part-2). Messages are routed round-robin across partitions by default.

Reliable Dispatch

Reliable dispatch persists messages before acknowledging sends. Use it for critical events that must not be lost.

let mut producer = client
    .new_producer()
    .with_topic("/default/critical-events")
    .with_name("reliable-producer")
    .with_reliable_dispatch()
    .build()?;

producer.create().await?;

reliableStrategy := danube.NewReliableDispatchStrategy()

producer, err := client.NewProducer().
    WithName("reliable-producer").
    WithTopic("/default/critical-events").
    WithDispatchStrategy(reliableStrategy).
    Build()

producer.Create(ctx)

Producer producer = client.newProducer()
        .withTopic("/default/critical-events")
        .withName("reliable-producer")
        .withDispatchStrategy(DispatchStrategy.RELIABLE)
        .build();

producer.create();

from danube import DispatchStrategy

producer = (
    client.new_producer()
    .with_topic("/default/critical-events")
    .with_name("reliable-producer")
    .with_dispatch_strategy(DispatchStrategy.RELIABLE)
    .build()
)

await producer.create()

See Dispatch Strategies for details on reliable vs. non-reliable delivery.

Schema Integration

Link producers to a schema subject for type-safe messaging:

let mut producer = client
    .new_producer()
    .with_topic("/default/events")
    .with_name("schema-producer")
    .with_schema_subject("event-schema")
    .build()?;

producer.create().await?;

producer, err := client.NewProducer().
    WithTopic("/default/events").
    WithName("schema-producer").
    WithSchemaSubject("event-schema").
    Build()

producer.Create(ctx)

Producer producer = client.newProducer()
        .withTopic("/default/events")
        .withName("schema-producer")
        .withSchemaLatest("event-schema")
        .build();

producer.create();

producer = (
    client.new_producer()
    .with_topic("/default/events")
    .with_name("schema-producer")
    .with_schema_subject("event-schema")
    .build()
)

await producer.create()

For schema registration and versioning, see the Schema Registry guide.

Topic Naming

Topics follow a namespace structure:

/namespace/topic-name

Examples:

/default/orders: orders in the default namespace
/production/user-events: user events in a production namespace
/staging/logs: logs in a staging namespace