Configuration Reference

All configuration options

Pipeline

Field	Behavior/Description	Required	Default	Type
name	name of the pipeline			string

License

Field	Behavior/Description	Required	Default	Type
license.acceptEula	It needs to be set to true to run the K2K application			boolean
license.token	A license key to enable premium features like exactly-once semantics [e-mail]			string

Feature Toggles

Required

Field	Behavior/Description	Required	Default	Type
features.exactlyOnce	Enables exactly-once processing.		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		enabled	enabled , disabled
features.checkKafkaConnectionOnStartup	At startup, if Kafka is not reachable, a set of checks is performed to understand the problem		disabled	enabled , disabled
features.headerReplication	Enables/disables header replication.		disabled	enabled , disabled
features.schemaMapping	Enables/disables schema replication.		disabled	enabled , disabled
features.optimizeOffsetCommitPartition	Optimizes offset commit messages by publishing all control messages to the same partition.		enabled	enabled, disabled
features.autoCreateControlTopics	Enables creation of the necessary control topics used by the k2k application.		disabled	enabled, disabled
features.autoCreateTopics	Enables automatic creation of replicated topics.		disabled	enabled, disabled
features.validateControlTopicSettings	Enables validation of configurations for control topics used by the k2k application.		enabled	enabled, disabled
features.tracingHeaders	Appends tracing headers to each replicated record. An object can be specified to configure which headers to provide.		disabled	enabled, disabled, object
features.tracingHeaders.partition	Appends the partition tracing header to each replicated record.		disabled	enabled, disabled
features.tracingHeaders.offset	Appends the partition tracing header to each replicated record.		disabled	enabled, disabled
features.tracingHeaders.topic	Appends the topic tracing header to each replicated record.		disabled	enabled, disabled
features.tracingHeaders.pipeline	Appends the pipeline name tracing header to each replicated record.		disabled	enabled, disabled
features.tracingHeaders.partition	Appends the partition tracing header to each replicated record.		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.		enabled	enabled, disabled
features.offsetMapping	Toggles the publishing of offset mapping records.		disabled	enabled, disabled
features.bidirectionalReplication	Global feature flag to enable bidirectional replication.		disabled	enabled, disabled

Tracing Headers

Field	Behavior/Description	Required	Default	Type
tracing.headers.partition	Name given to the header containing the source partition information.		__k2k_partition	string
tracing.headers.offset	Name given to the header containing the source offset information.		__k2k_offset	string
tracing.headers.topic	Name given to the header containing the source topic information.		__k2k_topic	string
tracing.headers.pipeline	Name given to the header containing the source pipeline information.		__k2k_pipeline	string

Error Handling

Field	Behavior/Description	Required	Default	Type
errorHandling.onCommitSyncTimeout	Handles the timeout when determining the latest committed offset.		fail	fail, ignore
errorHandling.onControlMessageDeserializationError	Handles deserialization errors for control messages.		fail	fail, ignore

Commit and Assignment

Required

Field	Behavior/Description	Required	Default	Type	Comment
coordination.assignment.topic	Topic used for assignment coordination.		__k2k-assignment	string	only used if exactly-once is enabled
coordination.assignment.graceWindow	Time to wait before the application actively fences off other applications.		15 seconds	duration string
coordination.assignment.fencingMaxParallelism	Controls the max parallelism when fencing slow/zombie producers.		10	integer
coordination.commit.topic	Topic for offset commits.		__k2k_consumer-offsets	string
coordination.commit.syncTimeout	Time limit to wait for catchup when reading from the offset control topic.		10 seconds	duration string
coordination.commit.batchSize	Number of records after which an offset is committed (when exactly-once is disabled). Batch size when exactly-once is enabled.		100	integer
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).		5 seconds	duration string

Bidirectional Replication Options

Fields in this section are only relevant if the bidirectional feature flag is enabled.

Field	Behavior/Description	Required	Default	Type
bidirectional.mode	Determines the identifier used in the bidirectional header. Can be either the pipeline name or the source cluster identifier.		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.		__k2k_bidirectional	string

Source Cluster Kafka and Schema Registry Connection

Only the connection details are required in this section (bootstrap servers and credentials). Those can be provided in the common section and will be inherited by the other Kafka client configurations targeting this cluster.

Fields targeting the Schema Registry are only relevant when the schema registry mapping feature is enabled.

Required

Field	Behavior/Description	Required	Default	Type
source.kafka.common	Common properties inherited by consumer and admin clients			object
source.kafka.consumer	Kafka consumer configuration. Used when reading data from the source cluster.			object
source.kafka.consumer."group.id"	Consumer group name. Used exclusively for partition assignment.			string
source.kafka.admin	Common properties inherited by consumer and admin and producer clients.			object
source.registry.config	Schema Registry connection options. (Full list of options in the corresponding section below).			object
source.registry.config."schema.registry.url"	Schema Registry URL.			string
source.registry.cacheSize	Schema Registry client internal client options. In most scenarios larger cache sizes won't result in improved performance.		"31"	numeric string
source.registry.headers	Headers specified here will be added to all schema registry HTTP requests.			object

Target Cluster Kafka and Schema Registry Connection

Fields targeting the Schema Registry are only relevant when the schema registry mapping feature is enabled.

Required

Field	Behavior/Description	Required	Default	Type
target.registry.common	Common properties inherited by consumer and admin and producer clients.			object
target.kafka.producer	Kafka producer configuration used to produce data to the target cluster.			object
target.kafka.admin	Kafka admin client configuration used to introspect the Kafka target cluster.			object
target.kafka.topicPropsExclusionList	Allows the exclusion of 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.			list of string
target.registry.config."schema.registry.url"	Schema Registry URL.
target.registry.config	Schema Registry connection options. (Full list of options in the corresponding section below).
target.registry.cacheSize	Schema Registry client internal client options. In most scenarios larger cache sizes won't result in improved performance.		"31"	numeric string
target.registry.headers	Headers specified here will be added to all schema registry HTTP requests.

Replication Pipeline

Refer to Source topics to get the full list of options that can be provided under replication[0].source.topic

For replication[1].sink.topic and replication[1].sink.partition refer to Records routing

Field	Behavior/Description	Required	Default	Type
replication[0].source.topic	Defines the topic(s) K2K will read data from.		string, object,list	cluster , pipeline`
replication[1].sink.topic	Strategy used to assign the target cluster's topic.		object, string	string
replication[1].sink.partition	Strategy used to assign the target cluster's partition for a record.			string

Schema Mapping Configuration

Fields targeting the Schema Registry are only relevant when the schema registry mapping feature is enabled.

Field	Behavior/Description	Required	Default	Type
schemaMapping.topics[].pattern	A regex that defines the topics schema mapping applies to.		regex	cluster , pipeline`
schemaMapping.topics[].key	Toggles schema replication for the topic's key facet.		enabled, disabled	enabled
schemaMapping.topics[].value	Toggles schema replication for the topic's value facet.		enabled, disabled	enabled

Metrics

Field	Behavior/Description	Required	Default	Type
metrics.prefix.targetConsumer	A regex that defines the topics schema mapping applies to.		k2k.consumer.control.	cluster , pipeline`
metrics.prefix.targetProducer	Prefix added to the target cluster producer metrics.		k2k.producer.	enabled
metrics.prefix.sourceConsumer	Prefix added to the source cluster consumer metrics.		k2k.consumer.source.	enabled
metrics.prefix.assignment	Prefix added to the K2K App assignment metrics.		k2k.	string
metrics.kafka.topic	Prefix added to the K2K App topic metrics.		__k2k_metrics	string

Control Topic Auto Creation

Only relevant if features.autoCreateControlTopics is enabled.

Required

Field	Behavior / Description	Required	Default	Type
topicCreation.control.common.partitions	Default partition count to be used when auto-creating any of the replicated topics.			integer
topicCreation.control.common.replication	Default replication to be used when auto-creating any of the replicated control topics.			short
topicCreation.control.common.config	Default topic configuration to be used when auto-creating any of the replicated control topics.		empty object	object
topicCreation.control.assignment.partitions	Regex used to select the topics the properties should be applied to. The first match in the containing array wins.			regex expression
topicCreation.control.assignemnt.replication	Partition count to be used when auto-creating replicated topics with a name that matches the specified pattern.			integer
topicCreation.control.assignemnt.config	Replication to be used when auto-creating replicated topics with a name that matches the specified pattern.		object with "cleanup.policy": "compact"	short
topicCreation.control.commit.partitions	Topic configuration to be used when auto-creating replicated topics with a name that matches the specified pattern			integer
topicCreation.control.commit.replication	Default replication to be used when creating the topic used for offset commit control.			integer
topicCreation.control.commit.config	Default topic configuration to be used when creating the topic used for offset commit control.		object with "cleanup.policy": "compact"	object
topicCreation.control.metrics.partitions	Default partition count to be used when creating the topic used for metrics.			integer
topicCreation.control.metrics.replication	Default replication to be used when creating the topic used for metrics.			integer
topicCreation.control.metrics.config	Default topic configuration to be used when creating the topic used for metrics.			object
topicCreation.control.offsetMapping.partitions	Default partition count to be used when auto-creating the offset-mappings topic.			integer
topicCreation.control.offsetMapping.replication	Default replication to be used when auto-creating the offset-mappings topic.			integer
topicCreation.control.offsetMapping.config	Default config to be used when auto-creating the offset-mappings topic.		object with "cleanup.policy": "delete"	object

Replicated Topics Auto Creation

Only relevant if features.autoCreateTopics is enabled.

Required

Field	Behavior / Description	Required	Default	Type
topicCreation.control.metrics.config	Default topic configuration to be used when creating the topic used for metrics.		empty	object
topicCreation.replication.common.partitions	Default partition count to be used when auto-creating any of the replicated topics.		empty	integer
topicCreation.replication.common.replication	Default replication to be used when auto-creating any of the replicated control topics.		empty	short
topicCreation.replication.common.config	Default topic configuration to be used when auto-creating any of the replicated control topics.		empty	object
topicCreation.replication.rules[].pattern	Regex used to select the topics the properties should be applied to. The first match in the containing array wins.			regex
topicCreation.replication.rules[].properties.partitions	Partition count to be used when auto-creating replicated topics with a name that matches the specified pattern.			integer
topicCreation.replication.rules[].properties.replication	Replication to be used when auto-creating replicated topics with a name that matches the specified pattern.			short
topicCreation.replication.rules[].properties.config	Topic configuration to be used when auto-creating replicated topics with a name that matches the specified pattern			object
license.acceptEula	It needs to be set to true to run the K2K application			boolean
license.token	A license key to enable premium features like exactly-once semantics [e-mail]			string

• If any of the properties topicCreation.*.replication and topicCreation.*.partitions is left blank (or set to null), K2K will default to the cluster default values for those properties.or control topics, the final topic configuration will be the result of merging the default/common value with the one specified in topicCreation.control.common.config and the one for the specific topic e.g: topicCreation.control.commit.config.
• For replicated topics, the final topic configuration will be the result of overriding the original topic's configuration with the result of merging the default/common value specified in topicCreation.replication.common with the one specified in the first matching rule in topicCreation.replication.rules[].

---

Registry config

The full list of Schema Registry client options and documentation can be found at: https://docs.confluent.io/platform/current/schema-registry/sr-client-configs.html#configuration-reference-for-sr-clients-on-product

Note: all options should be prefixed with schema.registry (see list below).

Field Name	Behavior
schema.registry.max.retries	Maximum number of client retries.
schema.registry.retries.wait.ms	Initial delay before first retry.
schema.registry.retries.max.wait.ms	Maximum delay between retries.
schema.registry.schema.registry.url.randomize	Randomize starting URL index for load balancing.
schema.registry.basic.auth.credentials.source	How to source Basic auth credentials (URL, USER_INFO, SASL_INHERIT).
schema.registry.basic.auth.user.info	User credentials for Basic auth in user:password.
schema.registry.http.connect.timeout.ms	HTTP connection timeout.
schema.registry.http.read.timeout.ms	HTTP read timeout.
schema.registry.bearer.auth.token	Static bearer token for authentication.
schema.registry.proxy.host	Hostname of HTTP proxy.
schema.registry.proxy.port	Port of HTTP proxy.
schema.registry.bearer.auth.credentials.source	Method to source bearer token (e.g., OAUTHBEARER).
schema.registry.bearer.auth.issuer.endpoint.url	OAuth/OIDC issuer token endpoint URL.
schema.registry.bearer.auth.client.id	OAuth client ID for client credentials grant.
schema.registry.bearer.auth.client.secret	OAuth client secret.
schema.registry.bearer.auth.scope	Requested OAuth token scope.
schema.registry.bearer.auth.scope.claim.name	JWT claim name for scope (default “scope”).
schema.registry.bearer.auth.sub.claim.name	JWT claim name for subject (default “sub”).
schema.registry.bearer.auth.logical.cluster	Logical cluster identifier for token header.
schema.registry.bearer.auth.identity.pool.id	Identity pool ID for token header.
schema.registry.bearer.auth.cache.expiry.buffer.seconds	Token cache expiry buffer before actual expiry.
schema.registry.bearer.auth.custom.provider.class	Class to implement custom token retrieval.
schema.registry.ssl.protocol	SSLProtocol (e.g. TLSv1.2, TLSv1.3).
schema.registry.ssl.provider	Java security provider name.
schema.registry.ssl.cipher.suites	Enabled SSL cipher suites.
schema.registry.ssl.enabled.protocols	Enabled SSL/TLS protocol versions list.
schema.registry.ssl.keystore.type	Keystore format (JKS, PKCS12).
schema.registry.ssl.keystore.key	Keystore private key.
schema.registry.ssl.keystore.certificate.chain	Certificate chain for keystore.
schema.registry.ssl.truststore.certificates	Trusted CA certificates.
schema.registry.ssl.keystore.location	Path to keystore file.
schema.registry.ssl.keystore.password	Password to access keystore.
schema.registry.ssl.key.password	Password for key inside keystore.
schema.registry.ssl.truststore.type	Truststore format.
schema.registry.ssl.truststore.location	Path to truststore file.
schema.registry.ssl.truststore.password	Password to access truststore.
schema.registry.ssl.keymanager.algorithm	KeyManagerFactory algorithm (e.g. SunX509).
schema.registry.ssl.trustmanager.algorithm	TrustManagerFactory algorithm (e.g. PKIX).
schema.registry.ssl.endpoint.identification.algorithm	Hostname verification algorithm (e.g. HTTPS).
schema.registry.ssl.secure.random.implementation	SecureRandom implementation class.
schema.registry.ssl.engine.factory.class	Custom SSLEngineFactory implementation.

---

Registry Supported Types

If your schema registry uses a format/type not listed as a default, you can add them here. That should be enough for K2K to be able to replicate your schemas.

---

(End of content)