--- title: "Produce events" slug: "dev-event-broker-guides-producing" description: "Learn to configure a Kafka producer for VAST Event Broker, covering settings, acks, keys, headers, batching, and common errors with Python examples." updated: 2026-10-30T18:18:46Z published: 2026-10-30T18:18:46Z canonical: "docs.vastdata.com/dev-event-broker-guides-producing" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Produce events *Applies to VAST Cluster 5.5.* This guide shows how to configure a Kafka producer for VAST Event Broker: the settings it needs, what `acks` means here, how keys pick a partition, headers, batching, and the errors you can get. The examples use the Confluent Kafka Python client (`confluent-kafka`). Run the code blocks in order in one Python session. Below each block is the output you should see. ## Required producer settings Every producer that writes to VAST Event Broker needs these settings: | Setting | Value | Why | | --- | --- | --- | | `enable.idempotence` | `false` | Idempotent producing isn't supported ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)). The Apache Kafka Java client turns idempotence on by default from version 3.0, so set it explicitly in every client. | | `compression.type` | `none` | Over-the-wire compression isn't supported. The broker rejects a compressed batch as an invalid message, and the client keeps retrying until the message times out, so a compressing producer reports `_MSG_TIMED_OUT`, not a compression error. The cluster compresses the stored data itself. | | `transactional.id` | not set | Transactions aren't supported. | This guide uses the Confluent Kafka Python client 2.4 to 2.8. Other supported clients are listed in [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility#supported-clients); the aiokafka client isn't supported ([5.5.0 release notes](/docs-5-5/docs/vast-cluster-5-5-0-release-notes)). If you use the kafka-python client instead, pin the protocol version on every admin client, producer and consumer with `api_version=(2, 0, 0)`. Without it, kafka-python 3.x tries the idempotent-producer request at start-up, which the broker doesn't offer. ``` from kafka import KafkaProducer producer = KafkaProducer(bootstrap_servers=BOOTSTRAP, api_version=(2, 0, 0), acks=1) ``` ## Set up You need the environment variables from the [quickstart](/develop/docs/dev-event-broker-quickstart#before-you-start). If authorization is turned on for the view, producing needs `KafkaWrite` on the topic; see [security](/develop/docs/dev-event-broker-operations-security). This block creates two topics: one with three partitions, and one with a small maximum batch size for the error examples. ``` import json import os from confluent_kafka import KafkaException, Producer, Consumer, TopicPartition from confluent_kafka.admin import AdminClient, NewTopic EVENTS = "docs-producing-events" SMALL = "docs-producing-small" def kafka_conf(extra=None): conf = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]} if os.environ.get("KAFKA_USERNAME"): conf.update({ "security.protocol": "SASL_PLAINTEXT", "sasl.mechanism": "PLAIN", "sasl.username": os.environ["KAFKA_USERNAME"], "sasl.password": os.environ["KAFKA_PASSWORD"], }) conf.update(extra or {}) return conf PRODUCER_SETTINGS = { "acks": "all", "enable.idempotence": False, "compression.type": "none", } def send(producer, topic, messages, timeout=30): """Produce (key, value, headers) tuples; return delivery results as (key, partition) or error names.""" results = [] def on_delivery(err, msg): key = msg.key().decode() if msg.key() else None results.append(err.name() if err else (key, msg.partition())) for key, value, headers in messages: producer.produce(topic, key=key, value=value, headers=headers, on_delivery=on_delivery) left = producer.flush(timeout) if left: results.append(f"{left} still queued") return results admin = AdminClient(kafka_conf()) for name, future in admin.create_topics([ NewTopic(EVENTS, num_partitions=3, replication_factor=1), NewTopic(SMALL, num_partitions=1, replication_factor=1, config={"max.message.bytes": "4096"}), ], operation_timeout=30).items(): future.result() print("created", name) ``` ``` created docs-producing-events created docs-producing-small ``` ## Produce with keys ![A producer sends events with keys ana, ben and chen, and some without a key, to a topic with three partitions. Events with the same key go to the same partition, so they keep their order. Events without a key are spread across the partitions. Different clients hash keys differently: the Confluent Python client uses CRC32 by default and the Java client uses murmur2.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/eb-keys-partitions.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A55%3A31Z&se=2026-11-08T09%3A55%3A31Z&sr=c&sp=r&sig=%2BUPIvmvCe%2FVIFyTPqFZ5ZlPOphk3e1vx%2FwyauoguleE%3D) *The key decides the partition, so events with the same key keep their order.* The key decides the partition. Events with the same key go to the same partition, so they keep their order. Events without a key are spread across the partitions: ``` producer = Producer(kafka_conf(PRODUCER_SETTINGS)) orders = [ (f"customer-{n % 4}", json.dumps({"order": n, "amount": 10 * n}), None) for n in range(12) ] results = send(producer, EVENTS, orders) partitions_per_key = {} for key, partition in results: partitions_per_key.setdefault(key, set()).add(partition) print("delivered:", len(results)) print("one partition per key:", all(len(p) == 1 for p in partitions_per_key.values())) ``` ``` delivered: 12 one partition per key: True ``` Kafka clients don't all hash keys the same way. The Confluent Python client uses CRC32 by default, and the Java client uses murmur2. If producers in different languages write to the same topic and you need a key to land in the same partition from all of them, set `"partitioner": "murmur2_random"` in the Python client. To write to a specific partition, pass `partition=`: ``` results = [] producer.produce(EVENTS, key="audit", value=json.dumps({"note": "pinned"}), partition=2, on_delivery=lambda err, msg: results.append(err.name() if err else msg.partition())) producer.flush(30) print("partition:", results) ``` ``` partition: [2] ``` ## Add headers Headers are key-value pairs that travel with the event. The values are bytes. Headers are stored with the event, and consumers and topic queries both see them. ``` [(_, partition)] = send(producer, EVENTS, [ ("customer-9", json.dumps({"order": 99}), [("source", b"web"), ("schema", b"order-v1")]), ]) reader = Consumer(kafka_conf({"group.id": "docs-producing-reader", "enable.auto.commit": False})) low, high = reader.get_watermark_offsets(TopicPartition(EVENTS, partition), timeout=10) reader.assign([TopicPartition(EVENTS, partition, low)]) for _ in range(30): # bounded: read until the event with headers turns up msg = reader.poll(1.0) if msg is not None and not msg.error() and msg.headers(): print(json.loads(msg.value()), msg.headers()) break reader.close() ``` ``` {'order': 99} [('source', b'web'), ('schema', b'order-v1')] ``` The reader uses `assign()`, not a consumer group subscription, so it reads the partition directly and doesn't commit offsets or create a group. See [consuming](/develop/docs/dev-event-broker-guides-consuming). ## Acknowledgements `acks` takes the usual Kafka values, with these meanings on VAST Event Broker: | `acks` | When the producer gets its reply | | --- | --- | | `all` (`-1`) | After the events are committed. Use this. | | `1` | The same as `all`. There are no follower replicas to wait for; the cluster stores each event once and protects it. | | `0` | Never: the broker doesn't reply, so the producer can't see errors. | Each Produce request is written as one VAST DataBase transaction, across all the partitions in the request. If writing any batch in the request fails, the whole request is rolled back, and every partition in it gets the error. Consumers and topic queries only see committed events. Any other `acks` value is rejected: ``` bad_acks = Producer(kafka_conf({**PRODUCER_SETTINGS, "acks": "2", "message.timeout.ms": 10000})) print(send(bad_acks, EVENTS, [("k", "v", None)], timeout=15)) ``` ``` ['INVALID_REQUIRED_ACKS'] ``` ## Delivery guarantees With idempotence off, a producer that retries after a lost reply can write the same events twice. Delivery is at least once. Make consumers tolerate duplicates, for example by using a business key to ignore events they've already processed, or use a compacted topic if only the latest value per key matters. See [topics](/develop/docs/dev-event-broker-guides-topics#compaction). ## Batching and throughput The producer groups events for the same partition into batches, and sends batches for several partitions in one request: - `linger.ms` is how long the producer waits to fill a batch (Confluent client default: 5 ms). A larger value gives fewer, larger requests. - `batch.size` and `batch.num.messages` cap a batch. - Each request is one database transaction, so fewer, larger requests mean fewer transactions. - A batch can't be larger than the topic's `max.message.bytes`. For benchmarking, producers can use `acks=1`, `compression.type=none` and `enable.idempotence=false`; see [Kafka performance benchmarking](https://kb.vastdata.com/docs/kafka-performance-benchmarking-with-openmessaging-framework). Measure with your own message sizes before you tune. ## What the broker rejects Show the errors for the common mistakes. Each producer has a short `message.timeout.ms` so the examples finish quickly. A compressed producer: ``` compressed = Producer(kafka_conf({**PRODUCER_SETTINGS, "compression.type": "gzip", "message.timeout.ms": 10000})) repetitive = json.dumps({"padding": "x" * 2000}) print(send(compressed, EVENTS, [("k", repetitive, None)], timeout=15)) ``` ``` ['_MSG_TIMED_OUT'] ``` The broker answers each compressed batch with `CORRUPT_MESSAGE` ("Invalid message"). The Confluent client treats that error as worth retrying, so it resends the batch until `message.timeout.ms` runs out, and the delivery report says `_MSG_TIMED_OUT`. If producers time out for no obvious reason, check `compression.type` first. An idempotent producer. The broker doesn't offer the request that a producer uses to get a producer ID, so the client reports the error itself: ``` client_errors = [] idempotent = Producer(kafka_conf({**PRODUCER_SETTINGS, "enable.idempotence": True, "message.timeout.ms": 10000, "error_cb": lambda err: client_errors.append(err.name())})) try: print("delivery:", send(idempotent, EVENTS, [("k", "v", None)], timeout=15)) except KafkaException as e: print(e.args[0].name(), "fatal:", e.args[0].fatal()) print(e.args[0].str()) print("client errors:", sorted(set(client_errors))) ``` ``` _UNSUPPORTED_FEATURE fatal: True Idempotent producer not supported by any of the 1 connected broker(s): requires Apache Kafka broker version >= 0.11.0 client errors: [] ``` A transactional producer. A transactional producer is always idempotent, so it needs `enable.idempotence=true`, and it fails the same way: ``` transactional = Producer(kafka_conf({**PRODUCER_SETTINGS, "enable.idempotence": True, "transactional.id": "docs-producing-tx"})) try: transactional.init_transactions(15) print("transactions initialised") except KafkaException as e: print(e.args[0].name(), "-", e.args[0].str()) ``` ``` _UNSUPPORTED_FEATURE - Transactions not supported by any of the 1 connected broker(s): requires Apache Kafka broker version >= 0.11.0 ``` A batch larger than the topic's `max.message.bytes` (4096 bytes on this topic): ``` print(send(producer, SMALL, [("k", "y" * 8192, None)])) ``` ``` ['MSG_SIZE_TOO_LARGE'] ``` A topic that doesn't exist. Topics aren't created automatically: ``` missing = Producer(kafka_conf({**PRODUCER_SETTINGS, "message.timeout.ms": 10000})) print(send(missing, "docs-producing-no-such-topic", [("k", "v", None)], timeout=15)) ``` ``` ['_MSG_TIMED_OUT'] ``` Other errors a producer can get: | Error | Cause | What to do | | --- | --- | --- | | `TOPIC_AUTHORIZATION_FAILED` | Authorization is on for the view, and the user doesn't have `KafkaWrite` on the topic. | Ask your platform team for the permission. See [security](/develop/docs/dev-event-broker-operations-security). | | `KAFKA_STORAGE_ERROR`, `NOT_LEADER_OR_FOLLOWER` | A temporary storage or routing problem. | Nothing: the client retries these. | | The event is rejected for its timestamp | With `message.timestamp.type=CreateTime`, the producer timestamp is outside the topic's allowed window. | Check the producer's clock, or the topic's timestamp settings. See [topics](/develop/docs/dev-event-broker-guides-topics#topic-settings). | | The connection closes with no error | The client sent a request or version that the broker doesn't support. | See [troubleshooting](/develop/docs/dev-event-broker-troubleshooting-index). | Producing isn't subject to VAST protocol auditing or QoS ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)). ## Clean up ``` producer.flush(30) for name, future in admin.delete_topics([EVENTS, SMALL], operation_timeout=30).items(): future.result() print("deleted", name) ``` ``` deleted docs-producing-events deleted docs-producing-small ``` ## See also - [Consuming](/develop/docs/dev-event-broker-guides-consuming): consumer groups, offsets and seeking. - [Topics](/develop/docs/dev-event-broker-guides-topics): partitions, retention, compaction and `max.message.bytes`. - [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility): supported requests and versions. - [Limits](/develop/docs/dev-event-broker-reference-limits): message, key and value sizes. - [Blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion): turn JSON values into table columns as you produce them. - [Troubleshooting](/develop/docs/dev-event-broker-troubleshooting-index).