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:

This guide uses the Confluent Kafka Python client 2.4 to 2.8. Other supported clients are listed in Kafka compatibility; the aiokafka client isn't supported (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. If authorization is turned on for the view, producing needs KafkaWrite on the topic; see 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

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.

Acknowledgements

acks takes the usual Kafka values, with these meanings on VAST Event Broker:

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.

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. 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:

Producing isn't subject to VAST protocol auditing or QoS (KB).

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: consumer groups, offsets and seeking.
• Topics: partitions, retention, compaction and max.message.bytes.
• Kafka compatibility: supported requests and versions.
• Limits: message, key and value sizes.
• Blob expansion: turn JSON values into table columns as you produce them.
• Troubleshooting.