Appearance

API reference@connectum/events-kafkaDocs 1.3.x

Connectum API Reference / @connectum/events-kafka / types / KafkaAdapterOptions

Interface: KafkaAdapterOptions ​

Defined in: types.ts:72

Options for creating a KafkaAdapter instance.

Properties ​

authenticationTimeout? ​

readonly optional authenticationTimeout?: number

Defined in: types.ts:89

Time allowed for the SASL handshake, in milliseconds.

brokers ​

readonly brokers: string[]

Defined in: types.ts:74

Kafka broker addresses (e.g., ["localhost:9092"])

clientId? ​

readonly optional clientId?: string

Defined in: types.ts:77

Client ID for this producer/consumer (default: "connectum")

clientOptions? ​

readonly optional clientOptions?: Readonly<Record<string, unknown>>

Defined in: types.ts:115

Settings of the underlying Kafka client that have no portable form, such as a custom logger factory, a socket factory or a custom SASL mechanism.

Not portable and not type-checked: the keys are those of the client library the adapter uses today (KafkaJS) and may change if that library is replaced. The value is merged shallowly over the typed options above (a retry here replaces retry there), and it can never override brokers or clientId.

connectionTimeout? ​

readonly optional connectionTimeout?: number

Defined in: types.ts:86

Time allowed to establish a connection, in milliseconds.

consumerOptions? ​

readonly optional consumerOptions?: object

Defined in: types.ts:127

Consumer-specific options

allowAutoTopicCreation? ​

readonly optional allowAutoTopicCreation?: boolean

Whether Kafka should auto-create topics on subscribe (default: false)

commitStrategy? ​

readonly optional commitStrategy?: "per-message" | "per-batch"

When the consumer group's offset is committed to the broker (default: "per-message").

  • "per-message": every acknowledged message is committed by its own OffsetCommit request, and await ack() returns once the broker has accepted it. A batch of 20 acknowledged messages costs 20 requests.
  • "per-batch": ack() only remembers the message; one OffsetCommit for the last acknowledged message is sent when the adapter stops working on the batch — at its end, at a requeued or unsettled message, when the handler throws, when the consumer stops, and when the group membership is lost. Acknowledged messages are never left uncommitted on any of these exits, and an unsettled message is never committed.

The price of "per-batch" is the window of duplicates. If the process dies (or the broker refuses the commit) between an ack() and the end of the batch, every message acknowledged in that batch is delivered again, up to the size of the batch, instead of at most one. Handlers must be idempotent, and a dead-letter copy published before the original is acknowledged can be published twice. await ack() no longer means the offset is durable. The delivery guarantee stays at-least-once.

KafkaAdapter() throws a RangeError for any other value.

fromBeginning? ​

readonly optional fromBeginning?: boolean

Whether to start consuming from the beginning of topics (default: false)

redeliveryDelay? ​

readonly optional redeliveryDelay?: number

Milliseconds to wait before a message that was not committed is delivered again (default: 1000).

A message stays uncommitted when the handler throws, calls nack() (requeue) or returns without settling it. Kafka offers no per-message redelivery timer, so the adapter pauses the affected partition for this long before the redelivery. 0 disables the pause: the same message is then fetched and handled again at network speed until it succeeds, is dead-lettered or is rejected with nack(false).

The pause only paces the redelivery loop; it does not end it. On an otherwise idle consumer the observed gap is a whole fetch cycle (5 s in KafkaJS) even for smaller values. Must be between 0 and 2147483647 (the longest delay a Node.js timer supports); KafkaAdapter() throws a RangeError otherwise.

sessionTimeout? ​

readonly optional sessionTimeout?: number

Session timeout in milliseconds (default: 30000)

topicDiscoveryInterval? ​

readonly optional topicDiscoveryInterval?: number | false

How often, in milliseconds, a subscription with wildcard patterns checks the broker for newly created topics that match them. Default: 300000 (five minutes, the metadata refresh period of the Kafka clients). false turns the checks off: the topic list is then fixed when subscribe() runs.

Without the checks a wildcard is expanded once, when subscribe() runs: a matching topic created later is not consumed until the service restarts, and a restart reads that topic from its end unless fromBeginning is set, so what was published to it before the restart is never handled by the group. A restart of the service inside one interval, before the check has seen a new topic, has the same effect for that topic: the window is at most one interval. With the checks, every interval the adapter lists the broker's topics and, if a matching topic appeared, restarts the subscription's consumer to include it. The restart rebalances the consumer group (a pause in consumption of a few seconds, in-flight messages are redelivered) and happens only when there is a new topic; the check itself is one metadata request per subscription per interval. A discovered topic is read from its first message, whatever fromBeginning says: it is new to the group, so nothing in it predates the subscription. A check that fails (for example the credentials may not list topics) is logged and repeated at the next interval. Each discovery is logged with the names of the topics, so a rebalance it causes can be told from one caused by a failing member.

In a group of several members every member must discover the topic before the group reads it completely: KafkaJS assigns partitions only from the topic list of the group's leader, and a member drops assigned topics it has not subscribed to itself. A new topic is therefore consumed in full after at most the longest interval among the members, with nothing lost in between. Give the members of a group the same interval.

The first check runs one interval after subscribe(). Subscriptions without wildcard patterns never check. A number must be positive, at most 2147483647; KafkaAdapter() throws a RangeError otherwise.

enforceRequestTimeout? ​

readonly optional enforceRequestTimeout?: boolean

Defined in: types.ts:98

Whether requestTimeout is enforced by the client itself rather than only by the socket.

lifecycle? ​

readonly optional lifecycle?: KafkaLifecycleCallbacks

Defined in: types.ts:118

Lifecycle callback. Without it the adapter attaches no listener to the client and behaves as before.

logLevel? ​

readonly optional logLevel?: KafkaLogLevel

Defined in: types.ts:104

Verbosity of the client library's log (default: the library's own default, "info").

producerOptions? ​

readonly optional producerOptions?: object

Defined in: types.ts:121

Producer-specific options

compression? ​

readonly optional compression?: KafkaCompression

Compression of produced messages (default: none).

reauthenticationThreshold? ​

readonly optional reauthenticationThreshold?: number

Defined in: types.ts:92

Share of a SASL session's lifetime after which the connection re-authenticates (a number between 0 and 1).

requestTimeout? ​

readonly optional requestTimeout?: number

Defined in: types.ts:95

Time after which a request without an answer fails, in milliseconds.

retry? ​

readonly optional retry?: KafkaRetryOptions

Defined in: types.ts:101

Retry of the client's requests and connection attempts.

sasl? ​

readonly optional sasl?: KafkaSaslOptions

Defined in: types.ts:83

SASL authentication of the broker connection.

ssl? ​

readonly optional ssl?: boolean | KafkaTlsOptions

Defined in: types.ts:80

TLS for the broker connection: true for the system trust store, or the settings of KafkaTlsOptions.