Skip to content
BoilStream

Produce with Kafka

tutorial
TL;DR

Use Kafka producers with Confluent Avro payloads and a table provisioned through SQL. Configure acknowledgements and idempotence explicitly: acks=1 confirms memory admission, while transactions and replication acknowledgements are unavailable. Authenticate Kafka and Schema Registry separately, check delivery reports and retain source records when outcomes are uncertain.

Message sequenceKafka producers receive an admission acknowledgement
Kafka producers receive an admission acknowledgementA producer encodes records with a compatible schema and sends a Produce request. BoilStream handles the request and returns an acknowledgement; this does not establish all Kafka broker durability guarantees.Kafka producerBoilStreamEncode records with schemaProduce request (acks=1)Validate and admit rowsProduce acknowledgementTime moves downward · dashed arrows are responsesKafka producers receive an admission acknowledgementA producer encodes records with a compatible schema and sends a Produce request. BoilStream handles the request and returns an acknowledgement; this does not establish all Kafka broker durability guarantees.KafkaproducerBoilStreamEncode records withschemaProduce request (acks=1)Validate and admit rowsProduce acknowledgementTime moves downward · dashed arrows are responses
An acknowledgement is an admission signal. Check supported protocol features and reconcile ambiguous writes before replaying.

Prerequisites

Section titled “Prerequisites”

Create a table in a streaming catalog and discover its Avro schema. Obtain the topic name, ingest token, separate registry token, endpoint and CA trust from your deployment.

Producer configuration

Section titled “Producer configuration”

Use the host, port, topic name, ingest token and trusted CA provided by your deployment. A Java producer configuration starts with:

Properties
bootstrap.servers=HOST:PORT
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="TOPIC" password="INGEST_TOKEN";
acks=1
enable.idempotence=false
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url=https://HOST/schema-registry
auto.register.schemas=false

The registry also requires a Schema Registry bearer token; configure it using your serializer client’s authentication options. It is separate from the Kafka ingest token. Configure the Kafka client’s TLS trust store and the registry’s TLS trust separately when using a private CA. Match the producer’s Avro schema to the registry schema and check every delivery callback. Client defaults vary, so set acknowledgements and idempotence explicitly.

For Python’s confluent-kafka, the corresponding producer settings are:

Python
producer_config = {
    'bootstrap.servers': 'HOST:PORT',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'PLAIN',
    'sasl.username': 'TOPIC',
    'sasl.password': 'INGEST_TOKEN',
    'acks': '1',
    'enable.idempotence': False,
}

Serialize values with Confluent Avro using the existing registry schema before producing. The dictionary above configures the connection; it does not serialize values. Use the client and serializer versions validated with your server release.

Verify delivery

Section titled “Verify delivery”

Inspect every per-partition delivery report and query the expected rows through PGWire after the commit becomes visible. Retain records after a lost response and reconcile before replay. See the protocol and acknowledgement reference.