Element <kafka-exporter>

Purpose: Declare a Kafka producer target used by generated product streams.

Why: Use it to publish generated messages to a configured Kafka topic.

Example

<kafka-exporter id="orders-out" topic="orders.generated"/>

Decision guide

Business value: Publishes generated records to a named Kafka topic through an explicit producer contract.

• Use when

  • A generated product must become a bounded stream of Kafka records.

• Choose another approach when

  • The output is a file artifact, database row, or RabbitMQ route.

• Prerequisites

  • Provide broker topology, topic, and deployment-owned security settings.

• Alternatives

  • Use rabbitmq-exporter for AMQP routing and publisher confirms. (See: <rabbitmq-exporter>)

Complete examples

Connect a bounded Kafka source to an artifact and topic target

Use separate importer and exporter ids so source consumption and outbound publication remain explicit contracts with independently selectable topics.

<setup>
    <kafka-importer id="orders_in"
                    system="kafkaLocal"
                    topic="{kafka_input_topic}"
                    group_id="{kafka_consumer_group}"
                    enable_auto_commit="false"
                    auto_offset_reset="earliest"
                    decoding="UTF-8"/>
    <kafka-exporter id="orders_out"
                    system="kafkaLocal"
                    topic="{kafka_output_topic}"/>
    <generate name="normalized_orders"
              source="orders_in"
              count="10"
              distribution="ordered"
              target="LogExporter,orders_out"/>
</setup>

Rules and invalid combinations

I349 — Export Kafka TLS Failed

Kafka TLS/SSL handshake failed for topic '{topic}': {reason}

Why: Kafka producer failed during TLS/SSL negotiation while exporting records.

Resolution: Verify Kafka TLS certificate paths, key material, and security protocol settings.

Full rule

I350 — Export Kafka Authentication Failed

Kafka authentication failed for topic '{topic}': {reason}

Why: Kafka producer credentials were rejected while exporting records.

Resolution: Verify SASL credentials/mechanism and retry with valid authentication settings.

Full rule

I351 — Export Kafka Authorization Failed

Kafka authorization failed for topic '{topic}': {reason}

Why: Kafka producer lacks permissions for the configured topic.

Resolution: Grant write permissions for the topic or update ACLs and retry.

Full rule

I352 — Export Kafka Network Failed

Kafka network connectivity failed for topic '{topic}': {reason}

Why: Kafka producer could not establish broker connectivity while exporting records.

Resolution: Verify broker hosts/ports, DNS, routing, and firewall configuration.

Full rule

I353 — Export Kafka Timeout

Kafka operation timed out for topic '{topic}': {reason}

Why: Kafka producer operation exceeded configured timeout while exporting records.

Resolution: Retry the operation or tune Kafka timeout settings for the environment.

Full rule

I354 — Export Kafka Topic Not Found

Kafka topic '{topic}' not found or inaccessible: {reason}

Why: Configured Kafka topic does not exist or cannot be accessed by the producer.

Resolution: Ensure the topic exists and the producer has access to it.

Full rule

I355 — Export Kafka Write Failed

Kafka write failed for topic '{topic}': {reason}

Why: Kafka producer encountered an unclassified write failure while exporting records.

Resolution: Inspect broker/client logs and producer configuration, then retry.

Full rule

Allowed parents / Allowed children

Allowed parents: else, else-if, if, setup, while

Allowed children:

None

Extension attributes: The declared attributes are complete. This element also accepts runtime-defined vendor extensions, which intentionally have no static completion.

Related concepts and use cases

• Choose a source and target — Guides users and agents from an input ownership boundary and intended side effect to one bounded source and target contract.

Attributes

Show all 51 attributes

acks

Broker acknowledgement policy: 0, 1, or all replicas.

optional; string; Default: null; Values: 0, 1, all.

allow_auto_create_topics

Allow the Kafka client to create missing topics when the broker permits it.

optional; boolean; Default: true.

api_version

Kafka protocol version as underscore-separated text or an integer tuple.

optional; string; Default: null.

api_version_auto_timeout_ms

Timeout in milliseconds for automatic broker-version detection.

optional; integer; Default: null.

batch_size

Maximum producer batch size in bytes.

optional; integer; Default: null.

bootstrap_servers

Kafka broker list (host:port).

optional; string.

buffer_memory

Total bytes available to buffer unsent producer records.

optional; integer; Default: null.

client_id

Client identifier reported to Kafka brokers.

optional; string; Default: null.

compression_type

Compression codec applied to producer batches.

optional; string; Default: null; Values: gzip, snappy, lz4, zstd.

connections_max_idle_ms

Idle time in milliseconds after which a broker connection is closed.

optional; integer; Default: null.

encoding

Optional encoding strategy.

optional; string; Default: null.

environment

Environment label for this exporter.

optional; string; Default: null.

format

Data format.

optional; string; Default: null; Values: string, json, avro.

id

Kafka exporter ID.

required; string.

linger_ms

Delay in milliseconds used to collect records into a producer batch.

optional; integer; Default: null.

max_block_ms

Maximum time in milliseconds a producer call may block for metadata or buffer space.

optional; integer; Default: null.

max_in_flight_requests_per_connection

Maximum number of unacknowledged requests per broker connection.

optional; integer; Default: null.

max_request_size

Maximum size in bytes of a producer request.

optional; integer; Default: null.

metadata_max_age_ms

Maximum age in milliseconds of cached broker metadata.

optional; integer; Default: null.

metrics_num_samples

Number of samples retained for Kafka client metrics.

optional; integer; Default: null.

metrics_sample_window_ms

Duration in milliseconds of each Kafka client metrics sample.

optional; integer; Default: null.

partition

Optional zero-based topic partition.

optional; integer; Default: null.

receive_buffer_bytes

TCP receive-buffer size in bytes; use the Kafka client default when omitted.

optional; integer; Default: null.

reconnect_backoff_max_ms

Maximum delay in milliseconds between broker reconnection attempts.

optional; integer; Default: null.

reconnect_backoff_ms

Initial delay in milliseconds before reconnecting to a broker.

optional; integer; Default: null.

request_timeout_ms

Maximum time in milliseconds to wait for a broker request.

optional; integer; Default: null.

retries

Maximum number of retries for a failed producer request.

optional; integer; Default: null.

retry_backoff_ms

Delay in milliseconds before retrying a failed broker operation.

optional; integer; Default: null.

sasl_kerberos_domain_name

Kerberos domain name used by GSSAPI authentication.

optional; string; Default: null.

sasl_kerberos_name

Kerberos principal name used by GSSAPI authentication.

optional; string; Default: null.

sasl_kerberos_service_name

Kerberos service name used by GSSAPI authentication.

optional; string; Default: null.

sasl_mechanism

SASL authentication mechanism.

optional; string; Default: null; Values: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.

sasl_plain_password

Password for PLAIN or SCRAM authentication.

optional; string; Default: null.

sasl_plain_username

Username for PLAIN or SCRAM authentication.

optional; string; Default: null.

schema

Schema/subject for schema-registry aware payloads.

optional; string; Default: null.

security_protocol

Transport and authentication protocol used for broker connections.

optional; string; Default: null; Values: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.

send_buffer_bytes

TCP send-buffer size in bytes; use the Kafka client default when omitted.

optional; integer; Default: null.

sock_chunk_buffer_count

Number of socket transfer chunks retained by the Kafka producer.

optional; integer; Default: null.

sock_chunk_bytes

Socket transfer chunk size in bytes used by the Kafka producer.

optional; integer; Default: null.

socket_options

Socket options as a Python literal string or a list accepted by the Kafka client.

optional; string; Default: null.

socks5_proxy

SOCKS5 proxy address used for broker connections.

optional; string; Default: null.

ssl_cafile

Path to the CA certificate file used to verify brokers.

optional; string; Default: null.

ssl_certfile

Path to the client certificate file.

optional; string; Default: null.

ssl_check_hostname

Verify that broker certificates match their host names.

optional; boolean; Default: null.

ssl_cipher_suites

OpenSSL cipher-suite expression used for broker connections.

optional; string; Default: null.

ssl_crlfile

Path to a certificate-revocation-list file.

optional; string; Default: null.

ssl_keyfile

Path to the client private-key file.

optional; string; Default: null.

ssl_password

Password used to decrypt the client private key.

optional; string; Default: null.

ssl_protocol

SSL protocol name passed to the Kafka client.

optional; string; Default: null.

system

System identifier for this exporter.

optional; string; Default: null.

topic

Kafka topic to publish to.

required; string.