Kafka connector
Reference: the kafka: block
Every forwarded Apache Kafka property the connector accepts, with types, defaults and bounds - plus the three lists of properties it refuses, and the capabilities no Kafka client can configure from a file.
Apache Kafka's own property names, forwarded. This is the half of the connector
configuration your existing client.properties knowledge transfers to.
Duraton's own keys are in the duraton: reference.
Values are parsed by declared type rather than guessed, so a number written as a string parses the same way. Enum values are case-insensitive. Every bound below is enforced at startup, not at connect time. Why an unknown name is fatal is What the connector refuses.
Connection
| Property | Type | Default | Bounds and notes |
|---|---|---|---|
bootstrap.servers | list | - | Required. A host with no port gets :9092. |
client.id | string | duraton-kafka-ingress | At most 256 bytes. |
client.rack | string | - | At most 512 bytes. |
request.timeout.ms | milliseconds | 10000 | 100 to 900000. |
socket.connection.setup.timeout.ms | milliseconds | 10000 | At least 1. |
connections.max.idle.ms | milliseconds | 30000 | 100 to 900000. |
metadata.max.age.ms | milliseconds | 300000 | 5000 to 3600000. |
retry.backoff.ms | milliseconds | 100 | At least 1, and no more than retry.backoff.max.ms. |
retry.backoff.max.ms | milliseconds | 1000 | At least 1. |
receive.message.max.bytes | bytes | 104857600 | 1024 to 1073741824. |
allow.auto.create.topics | boolean | false | Kafka defaults this on. A forwarder that creates a topic from a misspelling is worse than one that refuses to start. |
enable.metrics.push | boolean | true |
Consuming
| Property | Type | Default | Bounds and notes |
|---|---|---|---|
auto.offset.reset | enum | required | earliest, latest, none. Also accepts smallest and beginning for earliest, largest and end for latest, and error for none. |
isolation.level | enum | read_committed | read_committed, read_uncommitted. Kafka defaults to uncommitted; a forwarder cannot take back an event emitted from a transaction that later aborts. |
check.crcs | boolean | true | |
fetch.min.bytes | bytes | 1 | At least 1. |
fetch.max.bytes | bytes | 52428800 | No more than receive.message.max.bytes. |
fetch.max.wait.ms | milliseconds | 500 | At least 10. |
max.partition.fetch.bytes | bytes | 1048576 | No more than fetch.max.bytes. The combination is refused rather than quietly lowered. |
max.poll.records | integer | 500 | At least 1. Also checked against max.poll.interval.ms. |
Group membership
| Property | Type | Default | Bounds and notes |
|---|---|---|---|
group.id | string | - | Required unless duraton.assign is set, and mutually exclusive with it. |
group.instance.id | string | unset | Non-empty when present. Omit it to join as a dynamic member. |
partition.assignment.strategy | enum | cooperative-sticky | cooperative-sticky, sticky, range, roundrobin. A Java class name is refused with the short equivalent named. |
session.timeout.ms | milliseconds | 45000 | 100 to 2147483647. |
heartbeat.interval.ms | milliseconds | 3000 | At most a third of session.timeout.ms. |
max.poll.interval.ms | milliseconds | 300000 | 100 to 2147483647. |
Security
| Property | Type | Default | Notes |
|---|---|---|---|
security.protocol | enum | PLAINTEXT | PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. A SASL_* value requires sasl.mechanism. |
sasl.mechanism | enum | - | PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER. |
sasl.username | string | - | Required for PLAIN and both SCRAM mechanisms. |
sasl.password | secret | - | Never in the file. |
sasl.password.file | file path | - | |
sasl.oauthbearer.token.endpoint.url | string | - | OAUTHBEARER only. |
sasl.oauthbearer.client.credentials.client.id | string | - | OAUTHBEARER only. |
sasl.oauthbearer.client.credentials.client.secret | secret | - | Never in the file. OAUTHBEARER only. |
sasl.oauthbearer.client.credentials.client.secret.file | file path | - | |
sasl.oauthbearer.scope | string | - | OAUTHBEARER only. |
sasl.oauthbearer.extensions | list of key=value | - | OAUTHBEARER only. |
The TLS properties are in TLS and mTLS.
Alternative spellings
Each resolves to one canonical name. Writing both with different values is an error.
| Alternative | Canonical |
|---|---|
metadata.broker.list | bootstrap.servers |
topic.metadata.refresh.interval.ms | metadata.max.age.ms |
fetch.wait.max.ms | fetch.max.wait.ms |
fetch.message.max.bytes | max.partition.fetch.bytes |
sasl.mechanisms | sasl.mechanism |
sasl.oauthbearer.client.id | sasl.oauthbearer.client.credentials.client.id |
sasl.oauthbearer.client.secret | sasl.oauthbearer.client.credentials.client.secret |
socket.receive.buffer.bytes | receive.buffer.bytes (itself unsupported, below) |
socket.send.buffer.bytes | send.buffer.bytes (itself unsupported, below) |
Refused because they would break durability
Three, and only three. A property is controlled only where the connector's own correctness depends on it - never because a different value is merely preferred.
| Property | Why |
|---|---|
enable.auto.commit | Offsets would advance on a timer whether or not Duraton accepted the record, so a crash would silently drop events. The committed offset is the connector's only durability record. Writing false is accepted and changes nothing. |
auto.commit.interval.ms | Automatic commits are off, so there is no interval to set. Use duraton.commit.interval for how long an accepted record waits before its offset is committed. |
enable.auto.offset.store | Automatic commits are off, so a stored offset would never be committed. Offsets are stored explicitly once a record has been accepted. Unlike enable.auto.commit, no value is accepted here. |
Deliberately not controlled: session.timeout.ms, heartbeat.interval.ms, every fetch.*,
isolation.level, check.crcs, allow.auto.create.topics. Each has a default the connector chose and
can defend, and each is yours to override.
Refused because they cannot be honoured outside a JVM
The value is a Java class name, a Java keystore, or a login-module grammar, and there is no class loader.
| Property | The alternative |
|---|---|
sasl.jaas.config | sasl.username and sasl.password, or the OAUTHBEARER properties. |
ssl.truststore.location, ssl.truststore.password, ssl.truststore.type, ssl.truststore.certificates | ssl.ca.location or ssl.ca.pem. |
ssl.keystore.location, ssl.keystore.password, ssl.keystore.type, ssl.keystore.key, ssl.keystore.certificate.chain | ssl.certificate.location and ssl.key.location, or ssl.certificate.pem and ssl.key.pem. |
ssl.engine.factory.class, ssl.keymanager.algorithm, ssl.trustmanager.algorithm, ssl.secure.random.implementation, ssl.provider | The ssl.* properties the connector does accept. |
sasl.client.callback.handler.class, sasl.login.callback.handler.class, sasl.login.class | The OAUTHBEARER token provider in the duraton: block. |
interceptor.classes, metric.reporters, config.providers | None: these are Java class names. |
sasl.mechanism: GSSAPI and sasl.mechanism: AWS_MSK_IAM are refused for the same family of reasons,
each with its own message.
Refused because this connector cannot honour them
| Property | Why |
|---|---|
group.protocol | This connector does not select the consumer rebalance protocol. |
ssl.key.password | Encrypted private keys are not supported: the encryption scheme those files use is deprecated as insecure. Use an unencrypted key file protected by filesystem permissions. |
ssl.cipher.suites | The same property name carries OpenSSL cipher grammar in one Kafka client and Java suite names in another, and neither maps onto the suites this connector negotiates. Accepting it would silently mean something other than what was written. |
queued.min.messages, queued.max.messages.kbytes, fetch.queue.backoff.ms | There is no local prefetch queue to bound. The equivalent memory bound is fetch.max.bytes multiplied by the number of brokers. |
client.dns.lookup | Host names are resolved by the platform resolver, which offers no equivalent switch. |
default.api.timeout.ms, socket.timeout.ms | One request-deadline setting rather than three: use request.timeout.ms. |
reconnect.backoff.ms, reconnect.backoff.max.ms | One backoff covers retries and reconnects: use retry.backoff.ms and retry.backoff.max.ms. |
retries | A producer property in both Kafka vocabularies. The one record this connector ever writes is not separately tunable, and its consumer-side retry budget has no Kafka property, so nothing is lost. |
metadata.recovery.strategy, metadata.recovery.rebootstrap.trigger.ms | Seed brokers are already re-queried periodically, and the rebootstrap hook is a callback rather than a setting. |
receive.buffer.bytes, send.buffer.bytes | Socket buffer sizes are not exposed; the platform defaults apply. |
max.in.flight.requests.per.connection | Consumer fetches are already limited to one in flight per broker. |
exclude.internal.topics | Internal topics are never matched by a pattern subscription here. Use duraton.excludeTopics. |
enable.partition.eof | An end-of-partition signal has no meaning for a forwarder that never stops at the end of the log. |
api.version.request, api.version.fallback.ms, broker.version.fallback | All three are deprecated upstream and cover brokers older than 0.10.0. Broker API versions are negotiated automatically. |
share.acknowledgement.mode, share.acquire.mode | Share groups are deferred: their queue semantics overlap with what Duraton itself provides. |
sasl.kerberos.service.name, sasl.kerberos.principal, sasl.kerberos.keytab | GSSAPI is not built in this release, so its Kerberos settings have nothing to configure. |
Beyond passthrough
Some client capabilities take a function or an interface, and no Kafka client can configure them from a file - the Java equivalents are class names a JVM class loader resolves. This is the honest edge of "forwarding", so it is written down rather than left to be discovered.
| Capability | What the connector offers instead |
|---|---|
| Custom connection dialler | request.timeout.ms, socket.connection.setup.timeout.ms and the ssl.* properties. |
| Logger, context, hooks, buffer pools | Bound internally. Logging is configured in the duraton: block. |
| Retry backoff and retry timeout functions | Synthesised from retry.backoff.ms and retry.backoff.max.ms. |
| Rebootstrap hook | Seed brokers backed by DNS are re-resolved on reconnect. |
| Broker-pushed metrics callback | enable.metrics.push controls whether they are sent at all. |
| Preferring the laggiest partitions | No declarative equivalent. Can be added on request. |
| Custom decompressor | None needed: none, gzip, snappy, lz4 and zstd are built in, which is every codec Kafka defines. |
| Adjusting fetch offsets at assignment | No equivalent in this release. |
| Rebalance and offset callbacks | Bound internally to drain and commit before partitions are revoked, and surfaced as metrics and log events. |
| Auto-commit callback | Not applicable: automatic commits are off. |
| Share-group acknowledgement callback | Not applicable: share groups are off. |
| A TLS configuration object | Synthesised from the ssl.* properties. |
Next
duraton:reference - the other half of the file.- What the connector refuses - how to read a rejection.
- Security - the TLS properties and where secrets come from.
Reference: the duraton: block
Every key in the duraton block of the connector configuration, with its type, default and bounds - subscription, decoding, the envelope, the poison policy, retry windows, and the event mapping.
Worked examples
Two complete Kafka connector configurations with the workflows that consume them - order events given a lane per order, and call events threaded into one session across regional topics.