Kafka client configuration

StoatFlow constructs its own Kafka consumer, producer, and restoration consumers. You configure those clients by passing standard Kafka client properties through your application.yaml under stoatflow.kafka.*. This page covers the passthrough form, the opinionated defaults StoatFlow layers on top, and the small set of properties it forces for exactly-once correctness.

Passthrough form

Every property you set under stoatflow.kafka.consumer, stoatflow.kafka.producer, stoatflow.kafka.restoration-consumer or stoatflow.kafka.admin is passed straight to the corresponding Kafka client, using the exact Kafka property names (dotted keys):

stoatflow:
  bootstrap-servers: localhost:9092
  kafka:
    consumer:
      max.poll.records: 1000
      auto.offset.reset: latest   # StoatFlow defaults this to earliest; override to latest here
      fetch.min.bytes: 65536
    producer:
      batch.size: 524288
      linger.ms: 50
      compression.type: lz4

You don't need to list every property — only the ones you want to change. Anything you omit falls back to StoatFlow's framework default, and anything StoatFlow doesn't set falls back to the Kafka client default.

For where these keys sit in the wider configuration tree (and how YAML, environment variables, and system properties compose), see the configuration model and the core configuration reference.

How consumer properties resolve

Consumer properties are resolved by merging layers, each overwriting the previous:

Framework defaults

Forced overrides

These are always set by the framework. Setting them under stoatflow.kafka.consumer has no effect.

auto.offset.reset defaults to earliest (matching Kafka Streams) rather than the raw Kafka client's latest. It is not forced — override it globally under stoatflow.kafka.consumer, or per source topic with Consumed.withOffsetResetPolicy(...), which seeks explicitly and takes precedence for that topic.

One exception: with hot-standby HA enabled, group.instance.id is omitted (and any user-set value removed) — HA pods use assign() with no consumer-group membership, so static membership does not apply.

How producer properties resolve

Producer properties resolve through the same layered merge:

Framework defaults

Forced overrides

The barrier timeout is the stoatflow.commit-barrier.timeout-ms knob — see the core configuration reference.

Exactly-once mode only

When the processing guarantee is EXACTLY_ONCE, StoatFlow additionally forces the transactional producer properties:

In at-least-once mode (AT_LEAST_ONCE), enable.idempotence, acks, and max.in.flight.requests.per.connection stay at the framework-default layer and remain overridable via stoatflow.kafka.producer, so you can trade throughput against durability. In exactly-once mode they are forced: any value you set under stoatflow.kafka.producer is overwritten by the framework's forced override, so the client never sees a configuration that would break exactly-once. To see what the producer actually receives, check the resolved configuration (see Inspecting the resolved configuration below).

Restoration consumer

State stores recover from changelog topics on startup using dedicated restoration consumers, which are configured independently of the processing consumer under stoatflow.kafka.restoration-consumer. They inherit the processing stoatflow.kafka.consumer as a shared baseline, then layer restoration-specific defaults and overrides on top:

Framework defaults

Tuned for bulk changelog reads during recovery:

Forced overrides

Restoration tuning:

stoatflow:
  kafka:
    restoration-consumer:
      max.poll.records: 5000
      fetch.max.bytes: 104857600   # 100 MiB

Common tuning scenarios

High throughput

For maximum throughput with high event rates:

stoatflow:
  kafka:
    consumer:
      max.poll.records: 2000
      fetch.min.bytes: 65536        # 64 KiB — wait for larger fetches
      fetch.max.wait.ms: 200        # max wait for fetch.min.bytes
    producer:
      batch.size: 524288            # 512 KiB
      linger.ms: 50                 # allow more time to fill batches
      compression.type: lz4         # reduce network bandwidth

Low latency

For latency-sensitive workloads:

stoatflow:
  kafka:
    consumer:
      fetch.min.bytes: 1            # return immediately (Kafka default)
      max.poll.records: 100         # smaller batches for lower processing latency
    producer:
      linger.ms: 0                  # send immediately
      batch.size: 16384             # Kafka default (16 KiB)

Large records

When individual records are large (e.g. complex Avro > 10 KB):

stoatflow:
  kafka:
    consumer:
      max.partition.fetch.bytes: 20971520   # 20 MiB
      fetch.max.bytes: 104857600            # 100 MiB
    producer:
      max.request.size: 10485760            # 10 MiB
      buffer.memory: 536870912              # 512 MiB

Memory-constrained environments

When heap is limited:

stoatflow:
  kafka:
    consumer:
      max.poll.records: 100
      max.partition.fetch.bytes: 1048576    # 1 MiB (Kafka default)
      fetch.max.bytes: 10485760             # 10 MiB
    producer:
      batch.size: 32768                     # 32 KiB
      buffer.memory: 33554432               # 32 MiB (Kafka default)

Fast restoration

For large state stores where recovery time matters:

stoatflow:
  kafka:
    restoration-consumer:
      max.poll.records: 5000
      receive.buffer.bytes: 1048576         # 1 MiB socket buffer

Low-memory restoration

When restoration causes GC pressure on a constrained heap:

stoatflow:
  kafka:
    restoration-consumer:
      max.poll.records: 200
      max.partition.fetch.bytes: 1048576    # 1 MiB (Kafka default)
      fetch.max.bytes: 10485760             # 10 MiB

Setting properties via environment variables

Kafka client properties can also be set via environment variables, which take precedence over YAML. The format is STOATFLOW__KAFKA__{CONSUMER|PRODUCER|RESTORATION_CONSUMER|ADMIN}__{PROPERTY}, where the property name uses underscores instead of dots and is uppercased:

# Consumer properties
export STOATFLOW__KAFKA__CONSUMER__MAX_POLL_RECORDS=1000
export STOATFLOW__KAFKA__CONSUMER__FETCH_MAX_BYTES=104857600

# Producer properties
export STOATFLOW__KAFKA__PRODUCER__BATCH_SIZE=524288
export STOATFLOW__KAFKA__PRODUCER__LINGER_MS=50

# Restoration consumer properties
export STOATFLOW__KAFKA__RESTORATION_CONSUMER__MAX_POLL_RECORDS=5000

So max.poll.records becomes STOATFLOW__KAFKA__CONSUMER__MAX_POLL_RECORDS. See the configuration model for the full precedence rules.

SASL and SSL

Authentication and encryption are plain Kafka client properties — pass them through like any other:

Set them once, under consumer. StoatFlow propagates the consumer's security.* / ssl.* / sasl.* subset to every other client it builds — the producer, the admin clients (changelog topic management, broker health check, consumer-group preflight), the restoration consumers, and the HA metadata-log clients:

stoatflow:
  bootstrap-servers: broker-1:9093,broker-2:9093
  kafka:
    consumer:
      security.protocol: SASL_SSL
      sasl.mechanism: PLAIN
      sasl.jaas.config: ${KAFKA_SASL_JAAS_CONFIG}
      ssl.truststore.location: /etc/kafka/truststore.jks
      ssl.truststore.password: ${KAFKA_TRUSTSTORE_PASSWORD}

Keep secrets out of committed files — reference them with environment-variable placeholders (${...}) as shown.

If a particular client authenticates as a different principal, override it on that client's own map — stoatflow.kafka.producer, stoatflow.kafka.admin or stoatflow.kafka.restoration-consumer — which always wins over the inherited subset. Spell out every key of the family you are overriding: the override is per key, so anything you leave out is still inherited (see the warning above).

stoatflow:
  kafka:
    consumer:
      security.protocol: SASL_SSL
      sasl.mechanism: PLAIN
      sasl.jaas.config: ${KAFKA_SASL_JAAS_CONFIG}
    admin:
      # only when the admin principal differs from the consumer's
      sasl.jaas.config: ${KAFKA_ADMIN_SASL_JAAS_CONFIG}

Inspecting the resolved configuration

The merged application configuration is available at runtime from the /config HTTP endpoint, with sensitive values (passwords, secrets, tokens) masked. It serves JSON or YAML by content negotiation. See the REST API reference.

The resolved configuration is also logged at startup when the engine logger is at DEBUG:

logging:
  level:
    io.stoatflow.core.runtime.StreamProcessingEngine: DEBUG