Ctrlk
HomeBlogSign In
slack
For the complete documentation index, see llms.txt. This page is also available as Markdown.
BetaTeamEnterprise

Kafka

This page covers queue splitting for Kafka. For the general concepts and the message filter reference shared by all queue services, see the Queue Splitting overview.

The word "queue" on this page refers to a Kafka topic.

Queue splitting via MirrordSplitConfig requires mirrord operator 3.170.0 or later and mirrord CLI 3.221.0 or later.

⚠️ Deprecated CRD

MirrordKafkaTopicsConsumer + MirrordKafkaClientConfig are deprecated and replaced by MirrordSplitConfig. Existing resources continue to work for backward compatibility, but we recommend migrating to MirrordSplitConfig. See Migrating to MirrordSplitConfig.

The older operator.idleKafkaSplitTtlMillis Helm value (OPERATOR_KAFKA_SPLITTING_TTL) only affects legacy MirrordKafkaTopicsConsumer objects; with MirrordSplitConfig, use spec.drainTimeout instead.

How It Works

First, we have a consumer app reading messages from a Kafka queue:

A K8s application that consumes messages from a Kafka queue

When the first mirrord Kafka splitting session starts, two temporary queues are created (one for the target deployed in the cluster, one for the user's local application), and the mirrord operator routes messages according to the user's filter:

One Kafka splitting session

If a second user then starts a mirrord Kafka splitting session on the same queue, a third temporary queue is created (for the second user's local application). The mirrord operator includes the new queue and the second user's filter in the routing logic.

Two Kafka splitting sessions

If the filters defined by the two users both match some message, one of the users will receive the message at random.

Enabling Kafka Splitting in Your Cluster

1

Enable Kafka splitting in the Helm chart

Enable the operator.kafkaSplitting setting in the mirrord-operator Helm chart.

2

Configure the operator's Kafka client

The mirrord operator needs to be able to perform some operations on the Kafka cluster. The connection settings live in a MirrordPropertyList (CustomResource), which you reference from the MirrordSplitConfig (see the next step).

The MirrordPropertyList must live in the same namespace as the target workload (and the MirrordSplitConfig). Each property is a Kafka client property; the operator passes them straight to the underlying Kafka client. A few operator-specific keys (prefixed with mirrord.) are consumed by the operator and not forwarded to the client.

The full list of available Kafka client properties can be found here.

The operator recognizes these mirrord.-prefixed keys:

  • mirrord.client_implementation - the Kafka client backend, librdkafka (default) or java. Use java for Kafka Streams consumers (see below).

  • mirrord.auth.kind - extra authentication mechanism. The only supported value is MSK_IAM (see MSK IAM authentication).

  • mirrord.auth.aws_region - the AWS region, required when mirrord.auth.kind is MSK_IAM.

The Kafka consumer group used by the operator's own client is managed by mirrord, so a group.id property is not needed here.

If no MirrordPropertyList with the referenced name exists in the target's namespace, the operator falls back to a legacy MirrordKafkaClientConfig of the same name in the operator's namespace. This keeps older setups working untouched. See Migrating to MirrordSplitConfig.

3

Authorize deployed consumers

In order to be targeted with Kafka splitting, a deployed consumer must be able to use the temporary queues created by mirrord. E.g. if the consumer application describes the queue or reads messages from it — it must be able to do the same on a temporary queue. This might require extra actions on your side to adjust the authorization, for example based on queue name prefix. See customizing temporary queue names for more info.

4

Provide application context

On operator installation with operator.kafkaSplitting enabled, a new CustomResource type is defined in your cluster - MirrordSplitConfig. Users with permissions to get CRDs can verify its existence with kubectl get crd mirrordsplitconfigs.queues.mirrord.metalbear.co. Before you can run sessions with Kafka splitting, you must create a MirrordSplitConfig for the desired target. This tells the operator which queues to split and how the application discovers their names.

See an example MirrordSplitConfig for a meme app that consumes messages from a Kafka queue:

The MirrordSplitConfig above says that:

  1. It targets the deployment meme-app in namespace meme.

  2. The deployment consumes one queue. Its name is read from environment variable KAFKA_TOPIC_NAME in container consumer. If the variable is absent, the fallback value views-topic is used instead. The Kafka consumer group id is read from environment variable KAFKA_GROUP_ID in container consumer.

  3. The Kafka queue can be referenced in a mirrord config under ID views-topic.

  4. The Kafka client connection comes from the kafka-connection MirrordPropertyList.

Link the config to the deployed consumer

The MirrordSplitConfig is a namespaced resource, so it can only reference a consumer deployed in the same namespace. The target workload reference is specified with spec.targetRef:

  • apiVersion - API version of the Kubernetes workload (e.g. apps/v1, or argoproj.io/v1alpha1 for rollouts).

  • kind - type of the workload. The operator supports Kafka splitting on deployments, stateful sets, and Argo rollouts.

  • name - name of the workload.

Describe consumed queues

Each entry in the spec.queues list describes one or more Kafka queues consumed by the workload:

  • id - arbitrary queue ID that developers reference from their mirrord config.

  • kind - must be kafka.

  • clientConfig - name of the MirrordPropertyList with the Kafka client connection (from the previous step). Can also be set once for all Kafka queues with spec.clientConfigs.kafka.

  • appConfig.topic - how the application discovers the topic name. Each entry can use:

    • env - exact environment variable name containing the topic name.

    • envLike - regex matching environment variable names.

    • fallback - fallback topic name if the variable is absent (only valid with env). The env var is still rewritten to point at the temporary topic.

    • valueSelector - a jq expression to extract the topic name from the variable's value. Useful when the env var holds JSON rather than a plain name.

    • valuePattern - a regex used when the topic name is embedded in a larger string. The capture group (named value, otherwise the first group) marks the part that is the name; only that part is swapped for the temporary topic and the surrounding text is kept as-is.

    • containers - limit to specific containers (optional, defaults to all non-infra containers).

  • One of the following must be set:

    • appConfig.groupId - how the application discovers the consumer Kafka group id. Use for standard Kafka consumers. The operator's forwarder joins this group, and the consumer's environment is left untouched.

    • appConfig.appId - how the application discovers the Kafka Streams application id. Use for Kafka Streams consumers. The operator patches this variable to a fresh application id. Kafka Streams requires the Java client (mirrord.client_implementation: java).

    Both use the same structure as appConfig.topic.

The mirrord operator can only read consumer's environment variables if they are either:

  1. defined directly in the workload's pod template, with the value defined in value or in valueFrom via config map reference; or

  2. loaded from config maps using envFrom.

Additional Options

Customizing Temporary Kafka Queue Names

To serve Kafka splitting sessions, the mirrord operator creates temporary queues in the Kafka cluster. The default format for their names is as follows:

  • mirrord-tmp-1234567890-fallback-topic-original-topic - for the fallback queue (unfiltered messages, consumed by the deployed workload).

  • mirrord-tmp-0987654321-original-topic - for the user queues (filtered messages, consumed by local applications running with mirrord).

Note that the random characters will be unique for each temporary queue created by the operator.

You can adjust the format of the created queue names to suit your needs (RBAC, security, policies, etc.), using the OPERATOR_KAFKA_SPLITTING_TOPIC_FORMAT environment variable of the mirrord operator, or the operator.kafkaSplittingTopicFormat helm chart value. The default value is:

mirrord-tmp-{{RANDOM}}{{FALLBACK}}{{ORIGINAL_TOPIC}}

The provided format must contain the three variables: {{RANDOM}}, {{FALLBACK}} and {{ORIGINAL_TOPIC}}.

  • {{RANDOM}} will resolve to random characters.

  • {{FALLBACK}} will resolve either to -fallback- or - literal.

  • {{ORIGINAL_TOPIC}} will resolve to the name of the original topic that is being split.

Configuring Workload Restart

To inject the names of the temporary queues into the consumer workload, the operator always requires the workload to be restarted. Depending on cluster conditions, and the workload itself, this might take some time.

MirrordSplitConfig lets you tune this with two optional fields:

  • spec.restart.timeout - how long the operator waits for a new pod to become ready after the workload restart is triggered (in seconds, defaults to 60). This silences timeout errors when the workload pods take a long time to start.

  • spec.drainTimeout - how long the workload stays patched after its last Kafka splitting session ends (in seconds). While patched, a new session can reuse the split without another restart, and the workload can finish reading the temporary topic.

Two settings control the drain timeout:

Setting
Unit
Scope
Effect

spec.drainTimeout on the MirrordSplitConfig

seconds

One split

Wins over the cluster-wide default.

operator.kafkaSplittingDrainTimeout Helm value

milliseconds

Whole cluster

Default, used only when a config omits drainTimeout.

drainTimeout

Behavior

unset (both)

Unpatch as soon as the last session ends (same as 0). Messages not yet read from the temporary topic are lost.

0

Unpatch immediately. Messages not yet read from the temporary topic are lost.

N

Stay patched for up to N seconds so a new session can reuse the split, then unpatch.

AWS MSK IAM authentication

For Amazon Managed Streaming for Apache Kafka with IAM/OAUTHBEARER authentication, set the operator keys on the MirrordPropertyList:

When mirrord.auth.kind is MSK_IAM, the operator automatically adds sasl.mechanism=OAUTHBEARER and security.protocol=SASL_SSL.

To produce the authentication tokens, the operator uses the default credentials provider chain. The easiest way to provide the credentials is with IAM role assumption. For that, an IAM role with an appropriate policy has to be assigned to the operator's service account. Please follow AWS's documentation on how to do that. Note that the operator's service account can be annotated with the IAM role's ARN with the sa.roleArn setting in the mirrord-operator Helm chart.

Setting a filter

For the full filter reference (queue_type, message_filter, jq_filter), see the overview. Kafka uses queue_type: Kafka and supports message_filter on Kafka headers and jq_filter on a JSON representation of the whole record.

Filtering on headers

In the example above, the local application will receive a subset of messages from the Kafka queue with ID views-topic. All received messages will have a Kafka header baggage containing mirrord-session=alice.

Filtering on message content with jq

jq_filter runs a jq program on a JSON document the operator builds for each record:

  • topic - the topic name.

  • partition - the partition number.

  • offset - the record offset.

  • timestamp - the record timestamp in milliseconds (present only when the record carries one).

  • key - the record key (absent when the record has no key).

  • payload - the record value (absent when the record has no value).

  • headers - an object mapping header names to their values (always present, possibly empty; a repeated header name keeps the last value).

key, payload, and header values are UTF-8 strings, or base64-encoded when not valid UTF-8. A record matches if the jq program outputs true.

This lets you route messages by fields inside the message body. For example, to receive only messages whose JSON payload has a merchantId of 2137 under data:

If both message_filter and jq_filter are specified for the same queue, both must match for a message to reach the local application. Records for which the jq program errors (for example, a non-JSON payload piped to fromjson) are treated as not matching and stay on the deployed application's path.

jq_filter for Kafka requires mirrord operator 3.183.0 or later and mirrord CLI 3.232.0 or later, and is only supported with the default librdkafka client. Sessions using the Java client (mirrord.client_implementation: java, required for Kafka Streams) fail with a clear error when a jq_filter is set.

FAQ

How do I authenticate the operator's Kafka client with an SSL certificate?

Set the PEM contents as Kafka client properties on the MirrordPropertyList:

To avoid putting secrets directly in the resource, store them in a Kubernetes Secret and reference them with valueFrom:

By default, the mirrord operator has read access only to the secrets in the operator's namespace. The MirrordPropertyList itself lives in the target's namespace.

How do I authenticate the operator's Kafka client with a Java KeyStore?

The mirrord operator does not support direct use of JKS files. In order to use JKS files with Kafka splitting, first extract all necessary certificates and key to PEM files. You can do it like this:

Then, follow the guide for authenticating with an SSL certificate.

PreviousAmazon SQSNextRabbitMQ

Last updated

Was this helpful?