Configuration
Defaults are product choices for this client; see Compatibility
for the constructor-defaults table. Source:
types/index.ts.
KafkaConfig
| Field | Default | Notes |
|---|---|---|
brokers |
required | Bootstrap host:port, or () => string[] | Promise<string[]> |
ssl |
off | true or tls.ConnectionOptions. SSL |
sasl |
off | SASL. SCRAM also accepts tokenId / tokenHmac for delegation-token login |
clientId |
'' |
client.id |
connectionTimeout |
1000 |
Socket connect / TLS handshake, ms |
connectionsMaxIdleMs |
540000 |
Close a socket after this many ms idle (no send/receive, no in-flight requests). 0 disables. connections.max.idle.ms |
socketConnectionSetupTimeoutMaxMs |
30000 |
Cap for exponential growth of the connect timeout after consecutive failures. socket.connection.setup.timeout.max.ms |
clientDnsLookup |
'useAllDnsIps' |
'useAllDnsIps' resolves every A/AAAA and races them (RFC 8305 happy-eyeballs). 'canonicalBootstrap' follows CNAME/PTR for GSSAPI FQDN. client.dns.lookup |
reconnectBackoffMs |
50 |
Wait before reconnecting a dropped socket. 0 disables. reconnect.backoff.ms |
reconnectBackoffMaxMs |
1000 |
Cap for reconnect backoff. reconnect.backoff.max.ms |
authenticationTimeout |
SASL handshake, ms | |
reauthenticationThreshold |
Reauth before session expiry, ms | |
requestTimeout |
Per-request, ms | |
enforceRequestTimeout |
true |
|
metadataRecovery |
'rebootstrap' |
On REBOOTSTRAP_REQUIRED or an exhausted broker set, drop discovered metadata and reconnect to the original bootstrap list. 'none' keeps retrying known brokers. |
retry |
see below | |
logLevel |
logLevel.INFO |
Override with KAFKA_LOG_LEVEL |
logCreator |
console | Custom sink |
metrics |
off | true uses the global @opentelemetry/api meter (optional peer); { meter } supplies any compatible Meter. See Observability |
enableMetricsPush |
true |
KIP-714: subscribe and push client metrics to the broker when it advertises GetTelemetrySubscriptions (Kafka 3.5+). No-ops if the API is missing. enable.metrics.push |
Retry defaults (packages/core/src/retry/defaults.ts): retries: 5,
initialRetryTime: 300, maxRetryTime: 30000, multiplier: 2, factor: 0.2.
sasl.mechanism is plain, scram-sha-256, scram-sha-512, oauthbearer,
or gssapi. GSSAPI fields: serviceName (default kafka), optional
principal, keytab, krb5, authorizationIdentity, and gssProvider.
See Security.
ProducerConfig
| Field | Default | Notes |
|---|---|---|
idempotent |
false |
Explicit opt-in. enable.idempotence |
transactionalId |
transactional.id | |
transactionTimeout |
transaction.timeout.ms | |
acks |
-1 |
acks |
compression |
none | compression.type |
compressionLevel |
codec default | Honored by GZIP (zlib level, 0-9) and ZSTD (zlib.constants.ZSTD_c_compressionLevel, roughly 1-22). No-op for Snappy and LZ4 — see Throughput |
lingerMs |
5 |
Pass 0 for one Produce per send(). linger.ms |
batchSize |
16384 |
Pass 0 to not batch by size. batch.size |
createPartitioner |
murmur2 | Partitioners.StickyPartitioner adds opt-in KIP-794 sticky routing (enabled by throughputPreset()) |
metadataMaxAge |
300000 |
|
allowAutoTopicCreation |
true |
auto.create.topics.enable |
maxInFlightRequests |
5 |
Pass null to uncap. max.in.flight.requests.per.connection |
bufferMemory |
unset (unlimited) | Soft cap on linger-buffered bytes. The preset sets 32 MiB. buffer.memory |
retry |
5, or unlimited if idempotent | retries |
deliveryTimeoutMs |
120000 |
End-to-end deadline for one send/sendBatch call — lingerMs, any bufferMemory wait, and every retry, together. 0 disables it. delivery.timeout.ms |
maxRequestSize |
1048576 |
Cap, in bytes, on the uncompressed records of one Produce request. Enforced client-side before the broker ever sees the request. max.request.size |
hooks |
unset | Ordered async onSend/onAck hooks (not an interceptor SPI). See Producer hooks |
throughputPreset()
Named load profile. Does not change constructor defaults. Call it and spread:
import { throughputPreset } from '@cookiemonsterdev/kafka-core';
const { producer, consumer } = throughputPreset();
kafka.producer({ ...producer });
await kafka.consumer({ groupId }).run({
...consumer,
eachBatch: async ({ batch }) => {
for (const message of batch.messages) {
void message;
}
},
});
| Fragment | Fields |
|---|---|
producer |
sticky partitioner, bufferMemory 32 MiB (linger/batch/in-flight are constructor defaults) |
consumer |
partitionsConsumedConcurrently: 4 (a run() option) |
See Throughput and Compatibility.
ConsumerConfig
| Field | Default | Notes |
|---|---|---|
groupId |
required for subscribe() |
group.id. Optional for assign() (see Assign mode); needed there only to call commitOffsets |
groupProtocol |
'classic' |
group.protocol. 'consumer' opts into KIP-848 (Kafka 4.0+) |
sessionTimeout |
30000 |
session.timeout.ms. Unused when groupProtocol: 'consumer' |
rebalanceTimeout |
60000 |
max.poll.interval.ms |
heartbeatInterval |
3000 |
heartbeat.interval.ms. Unused when groupProtocol: 'consumer' |
partitionAssigners |
[roundRobin] |
Classic protocol only. KIP-848 uses server-side assignment |
groupRemoteAssignor |
group.remote.assignor. 'uniform' | 'range', KIP-848 only (groupProtocol: 'consumer'); ignored otherwise |
|
readUncommitted |
false |
Isolation defaults to read_committed. isolation.level |
autoOffsetReset |
auto.offset.reset. 'earliest', 'latest', 'none', or 'by_duration:<ISO-8601>' (KIP-1106) |
|
rackId |
'' |
client.rack |
groupInstanceId |
group.instance.id | |
maxBytesPerPartition |
1048576 |
max.partition.fetch.bytes |
minBytes |
1 |
fetch.min.bytes |
maxBytes |
10485760 |
fetch.max.bytes |
maxWaitTimeInMs |
5000 |
fetch.max.wait.ms |
checkCrcs |
true |
Verify each fetched batch’s CRC. false skips the check for throughput — see Throughput. check.crcs |
retry |
{ retries: 5 } |
|
hooks |
unset | Ordered async onConsume/onCommit hooks (not an interceptor SPI). See Consumer hooks |
committed, position, and currentLag on Consumer read committed
offsets, fetch position, and lag without any extra config. See
Consumer and
committed / position / currentLag.
ShareConsumerConfig
| Field | Default | Notes |
|---|---|---|
groupId |
required | Share group id (KIP-932) |
heartbeatInterval |
3000 |
Membership heartbeat interval, ms. The broker may override via ShareGroupHeartbeat |
maxWaitTimeInMs |
5000 |
ShareFetch max wait, ms |
minBytes |
1 |
ShareFetch min bytes |
maxBytes |
50MiB |
ShareFetch max bytes |
maxRecords |
500 |
ShareFetch max records |
batchSize |
100 |
ShareFetch batch size |
shareAcquireMode |
0 |
ShareFetch v2 (Kafka 4.2+): 0 batch-optimized, 1 record-limit (KIP-1206) |
rackId |
'' |
Optional rack for assignment |
retry |
{ retries: 5 } |
Requires Kafka 4.1+ with share groups enabled. See Consumer.
AdminConfig
| Field | Default | Notes |
|---|---|---|
retry |
inherited from KafkaConfig.retry |
|
bootstrapControllers |
unset | KIP-919: controller host:port list (or a function) instead of KafkaConfig.brokers for this admin instance. Requires DescribeCluster v1 (Kafka 3.7+). See Admin: Controller bootstrap |
Admin admin configs.