Skip to content

kafka://

Apache Kafka producer + consumer on top of github.com/segmentio/kafka-go. Implements ~30 commonly used URI parameters (offset management, fetch sizing, partitioner / partition-assignor, compression, retries, SASL+TLS security, OAuthBearer).

URI shape

kafka://<topic>?brokers=host:9092&groupId=group&autoOffsetReset=earliest&...
  • <topic> — topic name. With topicIsPattern=true it's treated as a regex matching multiple topics.
  • brokers — comma-separated host:port list. Falls back to the component default (localhost:9092).
  • A consumer endpoint requires groupId (or its legacy alias group); otherwise CreateConsumer returns ErrGroupRequiredForConsumer.

Parameters

The canonical list comes from (*Component).Describe() (37 functional params + the security stack). Grouped by role below.

Connection

Param Type Default Side Description
topic path — both Topic name (or pattern with topicIsPattern=true).
brokers string localhost:9092 both Comma-separated broker host:port list.
clientId string — both Client identifier sent on every broker request.
shutdownTimeout duration 30s both Max time Component.Stop waits for producers/consumers to drain.

Consumer

Param Type Default Description
groupId string — Consumer group ID. Required.
group string — Legacy alias for groupId.
autoOffsetReset enum latest earliest / latest / none / beginning / end.
seekTo enum — Alias: beginning ⇒ earliest, end ⇒ latest.
consumersCount integer 1 Parallel reader goroutines joined to the group.
pollTimeoutMs duration 5s Max wait per poll (Reader.MaxWait).
fetchWaitMaxMs duration 5s Alias for pollTimeoutMs.
fetchMinBytes integer 1 Min bytes per fetch.
fetchMaxBytes integer 52428800 Max bytes per fetch. Suffixes K/M/G accepted.
sessionTimeoutMs duration 45s Consumer-group session timeout.
heartbeatIntervalMs duration 3s Consumer-group heartbeat.
autoCommitEnable boolean true kafka-go auto-commits offsets every autoCommitIntervalMs.
autoCommitIntervalMs duration 5s Auto-commit cadence.
breakOnFirstError boolean false Stop the consumer on the first processor error.
topicIsPattern boolean false Treat the topic as a regex; lists topics from metadata and subscribes to matches.
consumerRequestTimeoutMs duration 30s Max wait between polls before back-off.
commitTimeoutMs duration 5s Informational (kafka-go auto-commits).
maxPollRecords integer 100 Records buffered per reader (ReaderConfig.QueueCapacity).
partitionAssignor enum range range / roundrobin. sticky / cooperative-sticky fall back to roundrobin with a WARN.
isolationLevel enum read_uncommitted read_uncommitted / read_committed.

Producer

Param Type Default Description
key string — Default message key (overridden by the kafka.KEY header).
partitionKey integer -1 Explicit partition number (-1 = let the partitioner decide).
partitioner enum hash hash / roundrobin / leastbytes / murmur2.
partitionerIgnoreKeys boolean false Ignore the key when partitioning (forces round-robin).
requestRequiredAcks enum all none / leader / all.
requestTimeoutMs duration 30s Per-write timeout (Writer.WriteTimeout).
producerBatchSize integer 16384 Batch size in bytes. Suffixes K/M/G accepted.
lingerMs duration 0 Wait for more messages before sending the batch (Writer.BatchTimeout).
compressionCodec enum none none / gzip / snappy / lz4 / zstd.
maxRequestSize integer 1048576 Max bytes per write (Writer.BatchBytes).
retries integer 10 Max write attempts (Writer.MaxAttempts).
retryBackoffMs duration 100ms Initial backoff (Writer.WriteBackoffMin).
retryBackoffMaxMs duration 1s Max backoff (Writer.WriteBackoffMax).

Security (SASL + TLS + OAuthBearer)

Param Type Default Description
securityProtocol enum PLAINTEXT PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL.
saslMechanism enum — PLAIN / SCRAM-SHA-256 / SCRAM-SHA-512 / OAUTHBEARER. Required when securityProtocol includes SASL.
saslUsername string — SASL username.
saslPassword string — SASL password.
tlsCAFile string — PEM CA bundle for broker cert verification.
tlsCertFile string — PEM client cert (mTLS).
tlsKeyFile string — PEM client key (mTLS).
tlsInsecureSkipVerify boolean false Skip server cert verification. Insecure.
tlsMinVersion enum — Pin minimum TLS version (TLSv1.2 / TLSv1.3).
tlsMaxVersion enum — Pin maximum TLS version.
tlsCipherSuites string — Comma-separated Go cipher-suite names.
oauthToken string — Pre-fetched OAUTHBEARER JWT (e.g. AWS MSK IAM). Mutually exclusive with client-credentials.
oauthClientId string — OAuth2 client ID for client-credentials grant.
oauthClientSecret string — OAuth2 client secret.
oauthScope string — OAuth2 scope (some IdPs require, many reject — leave empty if unsure).
oauthTokenEndpointUri string — OAuth2 token endpoint URL. Helix POSTs grant_type=client_credentials, caches the JWT until ~30 s before expiry.

Durations accept Go format (5s, 500ms) or bare integer milliseconds (5000).

Body shape

Producer

  • In body — sent verbatim as message Value (via BodyAsBytes()).
  • Reads kafka.KEY header (or key URI default) for the message key.

Consumer

Each polled message creates an exchange with:

  • In body — message Value (bytes).
  • In headers — kafka.KEY, kafka.TOPIC, kafka.PARTITION, kafka.OFFSET, kafka.TIMESTAMP.
  • fromUri — the endpoint URI.

Headers

Header Direction Description
kafka.KEY both Message key (producer overrides key URI param).
kafka.TOPIC inbound Source topic.
kafka.PARTITION inbound Source partition (as string).
kafka.OFFSET inbound Source offset (as string).
kafka.TIMESTAMP inbound Message time, RFC3339.

Examples

Minimal consumer:

- id: orders
  from: kafka://orders?brokers=kafka:9092&groupId=svc&autoOffsetReset=earliest
  steps:
    - to: postgres://orders/insert

Tuned consumer with multiple readers and read-committed isolation:

- id: events
  from: kafka://events?brokers=k1:9092,k2:9092&groupId=svc&consumersCount=8&isolationLevel=read_committed&maxPollRecords=500&pollTimeoutMs=2s
  steps:
    - to: direct://process-event

Producer with compression + acks=all:

- setHeader:
    name: kafka.KEY
    starlang: headers["customer-id"]
- to: kafka://customer-events?brokers=kafka:9092&compressionCodec=snappy&requestRequiredAcks=all&lingerMs=10ms

Topic pattern subscription:

- from: kafka://app\.(orders|payments)\..*?topicIsPattern=true&brokers=kafka:9092&groupId=svc

SASL/SCRAM over TLS:

- to: kafka://secure-topic?brokers=kafka.prod:9093&securityProtocol=SASL_SSL&saslMechanism=SCRAM-SHA-512&saslUsername=svc&saslPassword=<env:KAFKA_PASS>&tlsCAFile=/etc/ssl/kafka-ca.pem

AWS MSK IAM (externally-generated OAuthBearer token):

- to: kafka://my-topic?brokers=b-1.cluster.amazonaws.com:9098&securityProtocol=SASL_SSL&saslMechanism=OAUTHBEARER&oauthToken=<env:MSK_JWT>

Behavior notes

  • Default broker. localhost:9092 — override via brokers= or by calling kafka.NewWithBrokers(...) at startup.
  • Lazy writer init. Producer.Process initialises the kafka-go Writer on first message (sync.Once). The writer caches a Transport configured by the security stack.
  • breakOnFirstError. When true, the first processor error cancels the reader context and exits all reader goroutines. Inspect via Consumer.FirstError().
  • topicIsPattern. Resolves matching topics from broker metadata at consumer start. Topics added later won't be picked up until restart.
  • partitionAssignor=sticky isn't supported by kafka-go; falls back to roundrobin with a WARN.
  • partitionerIgnoreKeys=true forces round-robin even with hash set — useful when the message key is set for downstream consumption but you don't want partition affinity.
  • Suspend semantics. Consumer.Suspend() reads but drops the message (keeps the partition assignment). It does NOT pause the reader. To stop consuming, call Stop().
  • OAUTHBEARER caching. With client-credentials, Helix caches the JWT and refreshes ~30 s before expiry.

Source

  • pkg/component/kafka/kafka.go — Component, Endpoint, Producer, Consumer, parameter parsing, balancer / compression / isolation mapping.
  • pkg/component/kafka/security.go — SASL+TLS+OAuthBearer stack.
  • pkg/component/kafka/oauthbearer.go — client-credentials token cache.

See also