Element <kafka-importer>

Purpose: Declare a Kafka consumer source used by DataMimic statements.

Why: Use it as a bounded Kafka source for a generate or iterate statement.

Example

<kafka-importer id="orders-in" topic="orders.created"/>

Decision guide

Business value: Consumes a bounded set of Kafka records as a source for deterministic transformation.

• Use when

  • A model reads a topic with an explicit count or other supported bounded source contract.

• Choose another approach when

  • The requirement is an unbounded continuously running consumer.

• Prerequisites

  • Provide broker, topic, consumer-group policy, and a bounded consuming statement.

• Alternatives

  • Use rabbitmq-importer for bounded queue consumption. (See: <rabbitmq-importer>)

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

I159 — Source Kafka Client Missing

Kafka source requested but client was None

Why: Kafka source loading requires a configured Kafka client instance.

Resolution: Provide a valid Kafka client and retry.

Full rule

I915 — Generate Offset Unsupported for Kafka

'{stmt_name}': offset= is only supported for file sources, not Kafka source '{source_str}'.

Why: The offset attribute cannot be used with Kafka sources; it only applies to file-based sources.

Resolution: Remove the offset attribute, or use a file source instead of Kafka.

Full rule

I164 — Source Kafka Client Unsupported API

Kafka client '{client_type}' must implement consume_iter_message(count)

Why: The configured Kafka client does not implement the required consume API contract.

Resolution: Use a Kafka client adapter that implements consume_iter_message.

Full rule

I167 — Source Kafka TLS Failed

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

Why: The Kafka source client failed the TLS/SSL handshake while reading messages.

Resolution: Verify Kafka TLS certificate paths, trust chain, hostname validation, and protocol settings.

Full rule

I168 — Source Kafka Authentication Failed

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

Why: The Kafka source client could not authenticate with the configured credentials.

Resolution: Verify SASL/TLS identity settings and credentials for the Kafka source client.

Full rule

I169 — Source Kafka Authorization Failed

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

Why: The Kafka source client was authenticated but lacks permissions for the topic/resource.

Resolution: Grant required read permissions for the topic and related Kafka resources.

Full rule

I170 — Source Kafka Network Failed

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

Why: The Kafka source client could not reach the configured broker endpoint.

Resolution: Verify bootstrap server host/port, DNS, routing, and firewall connectivity.

Full rule

I171 — Source Kafka Timeout

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

Why: The Kafka source operation exceeded the configured timeout while reading messages.

Resolution: Retry or tune Kafka consumer timeout settings for this workload.

Full rule

I172 — Source Kafka Topic Not Found

Kafka topic/resource not found for topic '{topic}': {reason}

Why: The configured Kafka topic/resource was not found or is not accessible.

Resolution: Verify the topic exists, is spelled correctly, and is visible to the client.

Full rule

I173 — Source Kafka Read Failed

Kafka source read failed for topic '{topic}': {reason}

Why: The Kafka source client failed while reading messages due to an unexpected runtime error.

Resolution: Review Kafka client configuration and runtime logs, then retry the operation.

Full rule

I948 — Source Kafka Option Unsupported

Kafka source '{source}' does not support {option}={value}. Supported behavior: {supported}.

Why: The requested source-selection option cannot be preserved by bounded Kafka topic consumption.

Resolution: Remove the unsupported option or use the listed Kafka source behavior.

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.
• Determinism, parallelism, paging, and distribution — Separates seed, distribution, worker policy, paging, and streaming so topology choices do not become false determinism claims.

Attributes

Show all 56 attributes

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.

auto_commit_interval_ms

Interval in milliseconds between automatic offset commits.

optional; integer; Default: null.

auto_offset_reset

Policy used when the consumer group has no valid committed offset.

optional; string; Default: null; Values: latest, earliest, none.

bootstrap_servers

Broker list (host:port).

optional; string.

check_crcs

Validate CRC checksums of consumed records.

optional; boolean; Default: null.

client_id

Client identifier reported to Kafka brokers.

optional; string; Default: null.

connections_max_idle_ms

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

optional; integer; Default: null.

consumer_timeout_ms

Iterator timeout in milliseconds when no records are available.

optional; integer; Default: null.

decoding

Optional decoding strategy.

optional; string; Default: null.

enable_auto_commit

Automatically commit consumed offsets in the background.

optional; boolean; Default: null.

environment

Environment label for this importer.

optional; string; Default: null.

exclude_internal_topics

Exclude Kafka internal topics from subscription results.

optional; boolean; Default: null.

fetch_max_bytes

Maximum bytes returned by one consumer fetch request.

optional; integer; Default: null.

fetch_max_wait_ms

Maximum broker wait in milliseconds before returning a fetch response.

optional; integer; Default: null.

fetch_min_bytes

Minimum bytes a broker should collect before returning a fetch response.

optional; integer; Default: null.

format

Data format.

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

group_id

Consumer group identifier used for offset coordination.

optional; string; Default: null.

heartbeat_interval_ms

Interval in milliseconds between consumer-group heartbeats.

optional; integer; Default: null.

id

Kafka importer identifier.

required; string.

max_in_flight_requests_per_connection

Maximum number of unacknowledged requests per broker connection.

optional; integer; Default: null.

max_partition_fetch_bytes

Maximum bytes fetched from one partition per request.

optional; integer; Default: null.

max_poll_interval_ms

Maximum interval in milliseconds between consumer polls.

optional; integer; Default: null.

max_poll_records

Maximum records returned by one consumer poll.

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.

page_size

Maximum number of consumed records exposed as one DataMimic source page.

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.

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.

session_timeout_ms

Consumer-group session timeout in milliseconds.

optional; integer; 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 importer.

optional; string; Default: null.

topic

Topic to subscribe to.

required; string.