Docs / Kafka security and transactions

Kafka security and transactions

Authenticate with PLAIN or SCRAM, publish atomic transactions, and handle uncertain commit outcomes

The Kafka module supports TLS, SASL PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512. Its transactional producer publishes atomic groups of records across topics and partitions. Consumer offsets remain independent: this API does not provide an atomic consume-transform-produce transaction.

SASL configuration

Producer and consumer configurations use the same authentication settings. The settings also apply to reconnects, administrative requests, and retry/dead-letter publishers created by the Kafka transport.

config := viper.New()
config.Set("url", "kafkas://broker.example.com:9093")
config.Set("sasl_enabled", true)
config.Set("sasl_mechanism", "SCRAM-SHA-256")
config.Set("sasl_username", os.Getenv("KAFKA_USERNAME"))
config.Set("sasl_password", os.Getenv("KAFKA_PASSWORD"))

producer, err := kafka.NewKafkaProducer(config)
if err != nil {
    return err
}
defer producer.Close()
if err := producer.Start(ctx); err != nil {
    return err
}
return producer.Publish(ctx, "order-42", "orders", "customer-7",
    "application/json", []byte(`{"id":42}`))

Imports used in these fragments are github.com/spf13/viper, go.digitalxero.dev/mq-kafka/v2 as kafka, and the standard packages shown by usage. Put credentials into Viper explicitly or bind its environment variables; ${VARIABLE} inside YAML is not automatically expanded by mqutils.

Supplying SASL fields while sasl_enabled is false is a configuration error. SASL requires TLS unless sasl_allow_insecure is explicitly true. Enable TLS through kafkas:// or tls_enabled: true; skip_verify relaxes certificate verification without disabling encryption. Kafka uses the system TLS trust store; this adapter does not expose per-client CA files or client certificates. Validation errors do not include the configured username or password. OAuth and Kerberos/GSSAPI are outside this API.

The runnable SASL example accepts KAFKA_SASL_URL, KAFKA_SASL_MECHANISM, KAFKA_SASL_USERNAME, KAFKA_SASL_PASSWORD, and the development-only KAFKA_SASL_SKIP_VERIFY flag.

Atomic producer writes

Use kafka.NewTransactionalProducerBuilder(). Its Build(ctx) validates and connects the producer; a separate initial Start call is unnecessary. Keep a transaction ID stable for replacements of the same logical producer, and unique among simultaneously active producers. Kafka fences an older producer when a replacement uses its identity.

producer, err := kafka.NewTransactionalProducerBuilder().
    WithURL("kafkas://broker.example.com:9093").
    WithConfig(config). // the SASL settings above
    WithTransactionID("orders-writer-instance-1").
    WithTransactionTimeout(time.Minute).
    Build(ctx)
if err != nil {
    return err
}
defer producer.Close()

err = producer.InTransaction(ctx, func(tx types.Publisher) error {
    if err := tx.Publish(ctx, "order-42", "orders", "customer-7",
        "application/json", []byte(`{"id":42}`)); err != nil {
        return err
    }
    return tx.Publish(ctx, "order-42", "audit", "customer-7",
        "application/json", []byte(`{"event":"order.created","id":42}`))
})
if errors.Is(err, kafka.ErrTransactionOutcomeUnknown) {
    return fmt.Errorf("reconcile the order and audit publication before retrying: %w", err)
}
return err

The callback receives types.Publisher from go.digitalxero.dev/mqutils/v2/types. Returning nil commits only if all scoped publishes succeeded. An ignored publish error still prevents commit. A callback error, cancellation, or panic requests bounded abort/cleanup; a panic is rethrown. Finish every publish before returning from the callback.

The producer serializes entire transactions. Waiting callers can cancel. Ordinary Publish and PublishMsg on that producer return ErrTransactionRequired; only the callback publisher can publish. Do not recursively call InTransaction on the same producer. A retained callback publisher expires after the callback finishes.

tx.Close() invalidates that scope and requests abort. It never commits or closes the parent producer. Returning nil after closing the scope still yields ErrTransactionClosed; repeated Close is harmless, including after the scope expires. producer.Close() shuts down the owning producer.

transaction_timeout defaults to 1m and accepts 1ms through 15m, subject to the broker’s own transaction timeout limit. Transactions require Kafka 0.11 or newer, idempotent production, required_acks: all, and at least one producer retry; incompatible explicit settings are rejected. Idempotence prerequisites are configured by the transactional builder. Its WithConfig accepts the normal producer settings plus transaction_id and transaction_timeout. The ordinary producer rejects those transaction settings so they cannot be silently ignored.

Cancellation and uncertain outcomes

A transaction timeout bounds waiting for the callback and use of its publisher. It cannot stop arbitrary application code. The callback has no separate context parameter: pass a suitably bounded application context to your own work. Avoid irreversible external side effects inside a Kafka-only transaction.

A lost commit response can leave the broker outcome unknown. Check ErrTransactionOutcomeUnknown, reconcile durable application state, and do not blindly rerun the callback. A failed, fenced, or uncertain producer reports an unhealthy state and requires explicit Start(ctx) or a replacement producer before more transactions. Restarting a client does not decide whether an earlier transaction committed.

Read committed consumption

url: "kafka://localhost:9092"
destination: "orders"
consumer_group: "billing"
isolation_level: "read_committed"
handler: "process-order"

isolation_level accepts read_committed or read_uncommitted. The default remains read_uncommitted for compatibility. Read-committed consumers do not receive aborted transaction records and wait for transactions to resolve. Their own acknowledgment, retry-topic publication, and consumer-offset commits remain separate operations. Atomic producer writes do not make handler/database side effects exactly once.

The runnable transaction example writes an order and an audit event. Its -abort flag closes the scope, and it reports ambiguous outcomes separately. See the Kafka API reference for the complete interfaces and configuration.