| Field | Behavior / Description | Required | Default | Type | Comment |
|---|---|---|---|---|---|
name | Name of the pipeline. | Yes | string | ||
features.exactlyOnce | Enables exactly-once processing. | No | disabled | enabled, disabled | |
features.keepRecordCreationTimestamp | Keeps the original message timestamp. If disabled, a new timestamp is set based on the target topic og.message.timestamp.type setting | No | enabled | enabled, disabled | |
features.checkKafkaConnectionOnStartup | At startup, if Kafka is not reachable, a set of checks is performed to understand the problem | No | disabled | enabled, disabled | |
features.headerReplication | Enables/disables header replication. | No | disabled | enabled, disabled | |
features.schemaMapping | Enables/disables schema replication. | No | disabled | enabled, disabled | |
features.optimizeOffsetCommitPartition | Optimizes offset commit messages by publishing all control messages to the same partition. | No | enabled | enabled, disabled | |
features.autoCreateControlTopics | Enables creation of the necessary control topics used by the k2k application. | No | disabled | enabled, disabled | |
features.validateControlTopicSettings | Enables validation of configurations for control topics used by the k2k application. | No | enabled | enabled, disabled | |
features.tracingHeaders | Appends tracing headers to each replicated record. An object can be specified to configure which headers to provide. | No | disabled | enabled, disabled, object | |
features.tracingHeaders.partition | Appends the partition tracing header to each replicated record. | No | disabled | enabled, disabled | |
features.tracingHeaders.offset | Appends the offset tracing header to each replicated record. | No | disabled | enabled, disabled | |
features.tracingHeaders.topic | Appends the topic tracing header to each replicated record. | No | disabled | enabled, disabled | |
features.tracingHeaders.pipeline | Appends the pipeline name tracing header to each replicated record. | No | disabled | enabled, disabled | |
features.filterNonUpdatableTopicProperties | Toggles the filtering of properties that cannot be set, or are invalid in the target cluster when creating non-control topics. | No | enabled | enabled, disabled | |
features.bidirectionalReplication | Global feature flag to enable bidirectional replication. | No | disabled | enabled, disabled | |
tracing.headers.partition | Name given to the header containing the source partition information. | No | __k2k_partition | string | |
tracing.headers.offset | Name given to the header containing the source offset information. | No | __k2k_partition | string | |
tracing.headers.topic | Name given to the header containing the source topic information. | No | __k2k_partition | string | |
tracing.headers.pipeline | Name given to the header containing the source pipeline information. | No | __k2k_partition | string | |
errorHandling.onCommitSyncTimeout | Handles the timeout when determining the latest committed offset. | No | fail, ignore | ||
errorHandling.onControlMessageDeserializationError | Handles deserialization errors for control messages. | No | fail, ignore | ||
coordination.assignment.topic | Topic used for assignment coordination. | No | __k2k-assignment | only used if exactly-once is enabled | |
coordination.assignment.graceWindow | Time to wait before the application actively fences off other applications. | No | 15 seconds | Time string (e.g., 15 seconds) | |
coordination.assignment.fencingMaxParallelism | Controls the max parallelism when fencing slow/zombie producers. | No | 10 | integer | |
coordination.commit.topic | Topic for offset commits. | No | __k2k_consumer-offsets | ||
coordination.commit.syncTimeout | Time limit to wait for catchup when reading from the offset control topic. | No | 10 seconds | Time string (e.g., 10 seconds) | |
coordination.commit.batchSize | Number of records after which an offset is committed (when exactly-once is disabled). Batch size when exactly-once is enabled. | No | 100 | Positive integers | |
coordination.commit.batchTimeout | Time to wait before injecting a commit record (when exactly-once is disabled). Time to wait before committing if the batch size is not reached (when exactly-once is enabled). | No | 5 seconds | Time string (e.g., 5 seconds) | |
bidirectional.mode | Determines the identifier used in the bidirectional header. Can be either the pipeline name or the source cluster identifier. | No | cluster | cluster , pipeline` | |
bidirectional.headers.bidirectional | Name of the header used to mark bidirectional records. Checked when filtering out records that have already been replicated. | No | __k2k_bidirectional | string | |
source.kafka.consumer | Kafka consumer configuration. Used when reading data from the source cluster. | No | |||
source.kafka.common | Common properties inherited by consumer and admin clients | No | |||
source.kafka.consumer."group.id" | Consumer group name. Used exclusively for partition assignment. | Yes | This property is independent of coordination.kafka.commit.group | ||
source.kafka.admin | Common properties inherited by consumer and admin and producer clients. | No | |||
source.registry.config | Schema Registry connection options. (Full list of options in the corresponding section below). | No | empty object | object with string values. | only used if features.schemaMapping is enabled |
source.registry.config."schema.registry.url" | Schema Registry URL. | No | Comma-separated list of addresses. | only used if features.schemaMapping is enabled | |
source.registry.cacheSize | Schema Registry client internal client options. In most scenarios larger cache sizes won't result in improved performance. | No | 31 | Integer | only used if features.schemaMapping is enabled |
source.registry.headers | Headers specified here will be added to all schema registry HTTP requests. | No | empty object | object with primitive values | only used if features.schemaMapping is enabled |
target.registry.common | Common properties inherited by consumer and admin and producer clients. | No | |||
target.kafka.producer | Kafka producer configuration used to produce data to the target cluster. | No | |||
target.kafka.admin | Kafka admin client configuration used to introspect the Kafka target cluster. | No | |||
target.kafka.topicPropsExclusionList | allows you to exclude specified topic configurations when the replicator creates topics on the target cluster. The primary use case addresses compatibility issues with Kafka implementations that don't expose certain configurations as read-only in their metadata but reject them when included in topic creation requests. | No | empty array | Array of strings | |
target.registry.config."schema.registry.url" | Schema Registry URL. | No | Comma-separated list of addresses. | only used if features.schemaMapping is enabled | |
target.registry.config | Schema Registry connection options. (Full list of options in the corresponding section below). | No | empty object | object with string values. | only used if features.schemaMapping is enabled |
target.registry.cacheSize | Schema Registry client internal client options. In most scenarios larger cache sizes won't result in improved performance. | No | 31 | Integer | only used if features.schemaMapping is enabled |
target.registry.headers | Headers specified here will be added to all schema registry HTTP requests. | No | empty object | object with primitive values | only used if features.schemaMapping is enabled |
replication[0].source.topic | Defines the topic(s) K2K will read data from. | Yes | Section Replication.Source.Topic | ||
replication[].sink.topic | Strategy used to assign the target cluster's topic. | Yes | Any valid topic name | ||
replication[].sink.partition | Strategy used to assign the target cluster's partition for a record. | Yes | Any valid partition | ||
schemaMapping.topics[].pattern | A regex that defines the topics schema mapping applies to. | No | Regex | ||
schemaMapping.topics[].key | Toggles schema replication for the topic's key facet. | No | enabled/disabled | ||
schemaMapping.topics[].value | Toggles schema replication for the topic's value facet. | No | enabled/disabled | ||
metrics.prefix.targetConsumer | Prefix added to metrics concerning the consumer instantiated against the target cluster. | No | "k2k.consumer.control." | Any valid string | |
metrics.prefix.targetProducer | Prefix added to metrics concerning the producer instantiated against the target cluster. | No | "k2k.producer." | Any valid string | |
metrics.prefix.sourceConsumer | Prefix added to metrics concerning the consumer instantiated against the source cluster. | No | "k2k.producer." | Any valid string | |
metrics.prefix.assignment | Prefix added to metrics concerning the assignment process (assigned topics, partitions,...) | No | "k2k." | Any valid string | |
metrics.kafka.topic | Topic where metrics are published to. Note: OTEL_METRICS_EXPORTER must be configured with kafka for metrics to be exporter to kafka. | No | "k2k." | Any valid string | |
topicCreation.control.common.partitions | Default partition count to be used when creating any of the control topics. | No | empty | integer | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.common.replication | Default replication to be used when creating any of the control topics. | No | empty | short | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.common.config | Default topic configuration to be used when creating any of the control topics. | No | empty object | object | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.assignment.partitions | Default partition count to be used when creating the topic used for exactly-once assignment control. | No | empty | integer | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.assignment.replication | Default replication to be used when creating the topic used for exactly-once assignment control. | No | empty | short | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.assignment.config | Default topic configuration to be used when creating the topic used for exactly-once assignment control. | No | object with {"cleanup.policy": "compact, delete"} | object | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.commit.partitions | Default partition count to be used when creating the topic used for offset commit control. | No | empty | integer | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.commit.replication | Default replication to be used when creating the topic used for offset commit control. | No | empty | short | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.commit.config | Default topic configuration to be used when creating the topic used for offset commit control. | No | object with {"cleanup.policy": "compact, delete"} | object | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.metrics.partitions | Default partition count to be used when creating the topic used for metrics. | No | empty | integer | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.metrics.replication | Default replication to be used when creating the topic used for metrics. | No | empty | short | only used if features.autoCreateControlTopics is enabled |
topicCreation.control.metrics.config | Default topic configuration to be used when creating the topic used for metrics. | No | empty object | object | only used if features.autoCreateControlTopics is enabled |
topicCreation.replication.common.partitions | Default partition count to be used when auto-creating any of the replicated topics. | No | empty | integer | only used if features.autoCreateTopics is enabled |
topicCreation.replication.common.replication | Default replication to be used when auto-creating any of the replicated control topics. | No | empty | short | only used if features.autoCreateTopics is enabled |
topicCreation.replication.common.config | Default topic configuration to be used when auto-creating any of the replicated control topics. | No | empty object | object | only used if features.autoCreateTopics is enabled |
topicCreation.replication.rules[].pattern | Regex used to select the topics the properties should be applied to. The first match in the containing array wins. | No | regex expression | only used if features.autoCreateTopics is enabled | |
topicCreation.replication.rules[].properties.partitions | Partition count to be used when auto-creating replicated topics with a name that matches the specified pattern. | No | integer | only used if features.autoCreateTopics is enabled | |
topicCreation.replication.rules[].properties.replication | Replication to be used when auto-creating replicated topics with a name that matches the specified pattern. | No | short | only used if features.autoCreateTopics is enabled | |
topicCreation.replication.rules[].properties.config | Topic configuration to be used when auto-creating replicated topics with a name that matches the specified pattern | No | object | only used if features.autoCreateTopics is enabled | |
license.acceptEula | It needs to be set to true to run the K2K application | Yes | boolean | ||
license.token | A license key to enable premium features like exactly-once semantics k2k@lenses.io | No | string |