Kafka Producer ExecModule

Overview

KafkaProducerModule publishes one bounded text or JSON-compatible value to an Apache Kafka topic. Version 2.0 replaces the legacy simulator—which fabricated partitions and timestamp-based offsets without contacting Kafka—with the real Kafka producer client, acknowledgement metadata, bounded blocking time, IntegrationAccount-only SASL credentials, complete Workflow Studio metadata, and secret-safe failures.

The canonical module identifier is:

com.valkyrlabs.workflow.modules.integration.KafkaProducerModule

The module intentionally does not expose arbitrary producer properties, transactions, schema-registry integration, batch publishing, client-side DLQ claims, or automatic workflow retries.

Usage

1. Create a least-privilege Kafka principal with WRITE and DESCRIBE only for the required topic.
2. For SASL_SSL, create a READY IntegrationAccount: put the exact comma-separated broker set in accountId, the SASL username in username, and the SASL password in password.
3. Configure bootstrap_servers, topic, and message; optionally provide a record key and bounded string headers.
4. Select SASL_SSL, SSL, or an explicitly intended PLAINTEXT connection.
5. Execute the workflow and require status: success plus broker partition and offset before treating the publish as acknowledged.

Inputs

Input	Required	Description
bootstrap_servers	Yes	One through eight comma-separated host:port endpoints; no URLs or embedded credentials.
topic	Yes	Kafka topic with 1–249 safe characters.
message	Yes	Text or JSON-compatible value, serialized to at most 1 MiB UTF-8.
key	No	Optional record key, capped at 1 KiB UTF-8.
headers	No	Up to 32 string headers; each value is capped at 4 KiB and credential-like names are rejected.
security_protocol	No	SASL_SSL (default), SSL, or PLAINTEXT.
sasl_mechanism	No	PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.
compression	No	none, gzip, snappy, lz4, or zstd.
timeout_ms	No	Maximum block/delivery time from 1,000 through 60,000 ms. Default: 30,000.

Raw username, password, api_key, token, secret, and sasl_jaas_config fields are rejected. Legacy transactional_id, schema_registry_url, dlq_topic, and batch_mode claims are also rejected rather than silently simulated.

Outputs

Output	Description
status	success or error.
topic	Broker-acknowledged topic.
partition	Broker-selected partition.
offset	Broker-acknowledged record offset.
timestamp	Broker record timestamp when available.
serializedBytes	UTF-8 bytes submitted for the message value.
brokerIdempotence	Always true; describes Kafka client retries, not repeated workflow executions.
error	Safe object containing code, message, and retryable.

The module never returns SASL material, JAAS configuration, bootstrap credentials, arbitrary producer properties, raw provider exception text, or fabricated broker metadata.

IntegrationAccount Requirements

SASL_SSL requires a bound IntegrationAccount with:

• status exactly READY;
• accountId equal to the canonical bootstrap_servers set (order and whitespace do not matter, membership does);
• username containing the SASL username;
• password containing the SASL password;
• topic-scoped WRITE and DESCRIBE permissions only.

The exact broker binding prevents a workflow configuration change from redirecting credentials to a different Kafka cluster. SSL and PLAINTEXT do not read IntegrationAccount credentials; this module does not currently expose client-certificate configuration.

Configuration

Configuration	Default	Constraint
kafkaAccount	None	Required and READY for SASL_SSL.
bootstrap_servers	None	Required bounded broker set.
topic	None	Required safe topic.
message	None	Required, at most 1 MiB serialized.
key	None	Optional, at most 1 KiB.
headers	{}	Bounded string map without credential-like names.
security_protocol	SASL_SSL	SASL_SSL, SSL, or PLAINTEXT.
sasl_mechanism	PLAIN	PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.
compression	none	Kafka-supported allowlist.
client_id	valkyrai-kafka-producer	Non-secret safe identifier, at most 128 characters.
timeout_ms	30000	1,000–60,000 ms.

Operations

The module has one operation: publish one record and wait for Kafka acknowledgement.

It fixes these producer properties to preserve a narrow safety contract:

• acks=all;
• enable.idempotence=true;
• max.in.flight.requests.per.connection=5;
• bounded delivery.timeout.ms, request.timeout.ms, and max.block.ms;
• serializers fixed to UTF-8 string serializers;
• TLS hostname verification enabled for SSL and SASL_SSL;
• redirects, arbitrary interceptors, custom serializers, transaction IDs, and caller-defined retry settings unavailable.

Errors and Failure Modes

Code	Meaning	Recovery
VALIDATION_ERROR	A target, payload, header, protocol, compression, timeout, or forbidden field is invalid.	Correct the named field. No producer was created.
KAFKA_DELIVERY_FAILED	Account binding/authentication failed, the producer could not initialize, Kafka rejected the record, or acknowledgement timed out.	Check the account, broker TLS/SASL state, topic ACL, and cluster health. Reconcile possible delivery before retrying.
KAFKA_DELIVERY_INTERRUPTED	The workflow thread was interrupted while waiting for acknowledgement.	Reconcile the topic before retrying because delivery may be ambiguous.

Provider exception bodies are not surfaced because broker, TLS, or authentication errors may contain sensitive connection detail. A failure never includes a fabricated partition or offset.

Example

{
  "bootstrap_servers": "broker-1.example.com:9093,broker-2.example.com:9093",
  "topic": "orders.created",
  "message": {
    "orderId": "ord-42",
    "total": 125.00,
    "currency": "USD"
  },
  "key": "ord-42",
  "headers": {
    "event_type": "created",
    "schema_version": "1"
  },
  "security_protocol": "SASL_SSL",
  "sasl_mechanism": "SCRAM-SHA-512",
  "compression": "gzip",
  "timeout_ms": 30000
}

Expected shape after broker acknowledgement:

{
  "status": "success",
  "topic": "orders.created",
  "partition": 2,
  "offset": 9812,
  "timestamp": 1786279020000,
  "serializedBytes": 55,
  "brokerIdempotence": true
}

Notes

• Kafka producer idempotence protects eligible retries made by one producer session. It does not deduplicate a second workflow execution; consumers should still use business event IDs when end-to-end deduplication matters.
• The module performs one bounded send().get(...). On timeout or interruption, delivery can be ambiguous; reconcile the topic before retrying.
• There is no module-level retry loop. Kafka's idempotent producer may retry eligible broker errors within the configured delivery timeout.
• PLAINTEXT is available for explicitly trusted local/internal clusters and sends no credentials. Prefer SASL_SSL for production.
• Schema validation belongs before this module. The module serializes JSON-compatible values but does not contact a schema registry.
• The deterministic test suite uses Kafka's MockProducer, not a live broker. It verifies record construction, properties, SASL binding, validation, acknowledgement mapping, failure redaction, annotation discovery, and metadata serialization. Live TLS, SASL, broker ACL, compression codec, and cluster-version behavior remain deployment-time boundaries.