Loading
>
The skill file stays in the publisher's repository.
Paste the copied command in a terminal. It installs this SKILL.md into that agent.
npx skills add metalbear-co/mirrord --skill mirrord-kafka -a cursor -y
>
Which CRDs? Kafka splitting is now configured with MirrordSplitConfig (which queues to split + how the app finds their names) and MirrordPropertyList (the Kafka client connection). These replace the deprecated MirrordKafkaTopicsConsumer + MirrordKafkaClientConfig, which still work for backward compatibility. Generate the new resources for any new setup. Only produce the deprecated ones if the user explicitly asks or is maintaining an existing deployment. Requires operator 3.170.0+ and CLI 3.221.0+.
IMPORTANT: Follow these security rules for all operations in this skill.
MirrordPropertyList YAML. Reference a Kubernetes Secret with valueFrom.secretKeyRef per property.kubectl create secret generic ... --from-file=... reading values from files (then delete the files). Do not suggest --from-literal for credential values — it exposes secrets in argv/shell history.^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$ and reject shell metacharacters before interpolating into commands.kubectl get / kubectl config calls are read-only and safe. Never run kubectl apply/create/delete or helm install/upgrade on the user's behalf — present generated YAML and cluster-modifying commands for the user to review and run themselves.Guide DevOps engineers through the full setup of mirrord Operator's Kafka queue splitting:
operator.kafkaSplitting (and the Kafka sidecar for Kafka Streams)feature.split_queues section developers use to filter messages (message_filter on headers, jq_filter on record content)Step 1: Load reference files
references/mirrord-split-config-crd.md — MirrordSplitConfig field spec (current)references/mirrord-property-list-crd.md — MirrordPropertyList field spec, auth patterns (current)references/known-issues.md — active bugs, gotchas, and workaroundsreferences/kafka-topics-consumer-crd.md, references/kafka-client-config-crd.md — deprecated CRDs; read only when helping with an existing legacy setupAlways read the relevant CRD reference for any resource you generate.
Step 2: Inspect the cluster (if kubectl is available)
kubectl config current-context
kubectl cluster-info 2>/dev/null | head -5
# Operator present?
kubectl get ns mirrord --no-headers 2>/dev/null
kubectl get deploy mirrord-operator -n mirrord --no-headers 2>/dev/null
# Kafka splitting enabled? (current CRDs)
kubectl get crd mirrordsplitconfigs.queues.mirrord.metalbear.co --no-headers 2>/dev/null
kubectl get crd mirrordpropertylists.mirrord.metalbear.co --no-headers 2>/dev/null
# Existing configs
kubectl get mirrordsplitconfigs --all-namespaces --no-headers 2>/dev/null
kubectl get mirrordpropertylists --all-namespaces --no-headers 2>/dev/null
# Legacy CRDs (only if migrating an existing setup)
kubectl get crd mirrordkafkatopicsconsumers.queues.mirrord.metalbear.co --no-headers 2>/dev/null
If the operator is missing and the cluster has no mirrord for Teams license, an AI agent can offer the user a seven-day Enterprise trial and, once they agree, start it and install the operator: see the mirrord-operator skill, "Agent-started trial".
Inspect the target workload to extract container names and env vars:
kubectl get deployment/<name> -n <ns> -o yaml 2>/dev/null # or statefulset / rollout
kubectl get svc --all-namespaces --no-headers 2>/dev/null | grep -i kafka
This auto-discovery reduces the questions you need to ask (bootstrap server from a Kafka service; topic/group-id env vars from the target's pod spec). If kubectl isn't available, ask.
Step 3: Gather remaining context
For MirrordPropertyList:
For MirrordSplitConfig:
MirrordPropertyList name to referenceRemind the user once, early, to enable Kafka splitting:
operator:
kafkaSplitting: true
# For Kafka Streams consumers only:
kafkaSplittingSidecar:
enabled: true
Rules:
MirrordSplitConfig) — this is the recommended primary location for a single team's connection config, and it wins if a list of the same name also exists in the operator's namespace. The operator (3.191.0+) also looks up the list in its own namespace as a fallback, so one connection config can be shared across many teams/namespaces — only reach for that when the user explicitly wants shared/cluster-wide credentials. ConfigMap/Secret refs inside the list resolve in whichever namespace the list itself was found in.group.id — mirrord manages the operator's consumer group.INCONSISTENT_GROUP_PROTOCOL: set mirrord.temporary_group_id: "true" (operator 3.195.0+). This is different from the Kafka Streams case below — it's for regular consumers whose client library advertises a custom partition-assignment protocol the operator's librdkafka consumer can't join.PolicyViolation (e.g. Confluent Cloud requires replication factor 3): set mirrord.split_topic.replication_factor (operator 3.191.0+) to a positive number, copy (match the source topic's factor), or -1 (broker default).valueFrom.secretKeyRef for any credential (SASL password, SSL PEMs, key password).mirrord.auth.kind: MSK_IAM + mirrord.auth.aws_region (auto-adds OAUTHBEARER + SASL_SSL).mirrord.client_implementation: java.security.protocol to SASL_SSL when the user mentions SASL without specifying transport, and flag it: "defaulted to SASL_SSL — change to SASL_PLAINTEXT if your broker uses plaintext transport."apiVersion: mirrord.metalbear.co/v1
kind: MirrordPropertyList
metadata:
name: kafka-connection
namespace: <target-namespace>
spec:
properties:
- name: bootstrap.servers
value: <broker-address>
- name: security.protocol
value: PLAINTEXT
# credentials via valueFrom.secretKeyRef, MSK IAM keys, or client_implementation as needed
See references/mirrord-property-list-crd.md for MSK IAM, SSL-via-Secret, Streams, and Java KeyStore credentials (native mirrord.ssl.*.base64 on operator 3.199.0+, JKS→PEM conversion for older operators).
Rules:
spec.targetRef = { apiVersion, kind, name } (Deployment/StatefulSet/Rollout).spec.queues[] needs id, kind: kafka, a clientConfig (the MirrordPropertyList name; or set once via spec.clientConfigs.kafka), and appConfig.topic.appConfig.groupId (standard consumers) or appConfig.appId (Kafka Streams) per queue.spec.restart.timeout (pod readiness wait after a restart), spec.ttl (idle window: keeps the split fully live so a reconnecting session resumes instantly, requires operator 3.194.0+), and spec.drainTimeout (drain window that follows: lets the workload finish the already-forwarded backlog before unpatching). On operators older than 3.194.0, spec.drainTimeout alone controls how long the workload stays patched after the last session.mirrord.temporary_group_id) waits for the workload's rollout to finish — 180 seconds by default, then the session fails. For a slow rollout (many replicas, a long termination grace period, a consumer that stays in the group until its session timeout expires), raise it with mirrord.group_join_timeout (seconds) on the MirrordPropertyList (operator 3.204.0+; older operators reject it as an unknown mirrord. key).apiVersion: queues.mirrord.metalbear.co/v1
kind: MirrordSplitConfig
metadata:
name: <workload>-split
namespace: <target-namespace>
spec:
targetRef:
apiVersion: apps/v1
kind: Deployment
name: <workload-name>
queues:
- id: <topic-id>
kind: kafka
clientConfig: kafka-connection
appConfig:
topic:
- env: <TOPIC_ENV_VAR>
fallback: <topic-name> # optional
containers: [<container>]
groupId:
- env: <GROUP_ID_ENV_VAR>
containers: [<container>]
appConfig.topic/groupId/appId sources also support envLike (regex over var names), volume (read the name from a file mounted from a ConfigMap volume instead of an env var — requires operator 3.198.0+; see references/mirrord-split-config-crd.md), podFile (read the name from a file that exists only inside the running pods — e.g. rendered by vault-agent-injector or a secrets-store CSI driver, with no ConfigMap/Secret behind it — requires operator 3.201.0+; see below), valueSelector (a selector over nested keys / .[] for JSON-valued env vars — not a full jq expression, no pipes or functions), and valuePattern (regex to swap an embedded name). See the split-config reference.
podFile source (Vault/CSI-injected names): the operator reads the file by running cat in a running pod of the target, so the target needs at least one running pod when the split starts, and the operator needs get/create on pods/exec in the target namespace (the Helm chart grants this when Kafka splitting is enabled). It then mounts a Secret carrying the substituted content over the file's exact path in the app containers — the same kind of restart env-var injection causes — while the injector's own sidecar keeps rendering the original underneath. podFile.path is the absolute in-container path; podFile.container defaults to a vault-agent sidecar if present, else the pod's first app container (set it explicitly if the default container has no cat, e.g. distroless). If both podFile and an env/envLike/volume source are set on the same entry, the other source wins and podFile is ignored; fallback doesn't apply to it. The referenced file's content is pinned for the split's duration — a value that also rotates (like a credential) keeps reading the value from split start.
Show the developer-facing config referencing the topic IDs. Two filter kinds, and you can combine them:
Filter on Kafka headers (message_filter):
{
"operator": true,
"target": "deployment/<workload>/container/<container>",
"feature": {
"split_queues": {
"<topic-id>": {
"queue_type": "Kafka",
"message_filter": { "<header-name>": "<regex>" }
}
}
}
}
All specified headers must match. An empty message_filter: {} with no jq_filter is match-none (the local app gets zero messages).
Composable header filter (filter) — NEW, alternative to message_filter:
{
"operator": true,
"target": "deployment/<workload>/container/<container>",
"feature": {
"split_queues": {
"<topic-id>": {
"queue_type": "Kafka",
"filter": {
"all_of": [
{ "metadata": "^tenant: blue$" },
{ "metadata": "^region: eu-.*$" }
]
}
}
}
}
}
filter takes a single { "metadata": "<regex>" }, or an any_of/all_of list of them. Each metadata regex is matched against every header rendered as <name>: <value> — one regex can pin a header by name or match a marker wherever it's propagated. A message_filter of {"tenant": "^blue$"} is equivalent to filter: {"metadata": "^tenant: blue$"}, except message_filter requires the header name to match exactly while a metadata regex sees the whole name: value line. Use either filter or message_filter on an entry, not both. Requires mirrord 3.264.0+ and operator 3.212.0+. A metadata regex can't be verified against a specific header name, so a topic covered by a splitQueues policy rule rejects a lone metadata filter the same way it rejects a lone jq_filter — use message_filter there instead.
Filter on record content (jq_filter) — NEW:
{
"operator": true,
"target": "deployment/<workload>/container/<container>",
"feature": {
"split_queues": {
"<topic-id>": {
"queue_type": "Kafka",
"jq_filter": ".payload | fromjson | .data.merchantId == 2137"
}
}
}
}
jq_filter runs a jq program over a JSON doc the operator builds per record: topic, partition, offset, timestamp, key, payload, headers. key/payload/header values are UTF-8 strings (or base64 when not valid UTF-8). A record matches if the program outputs true; a record whose program errors (e.g. fromjson on non-JSON) is treated as not matching and stays on the deployed app's path.
Filter on protobuf payloads (payload_protobuf) — NEW:
{
"operator": true,
"target": "deployment/<workload>/container/<container>",
"feature": {
"split_queues": {
"<topic-id>": {
"queue_type": "Kafka",
"payload_protobuf": {
"schema_file": "schemas/cdc_record.proto",
"message_type": "com.example.cdc.Record"
},
"jq_filter": ".payload_decoded.merchant_id == 2137"
}
}
}
}
For topics carrying raw protobuf record values (no JSON envelope, no schema-registry framing) instead of JSON. schema_file points at a local .proto file the CLI compiles itself (resolving imports against the file's directory — add include_directories for extra import roots); message_type is the fully-qualified message type. Users with a pre-compiled schema can set descriptor_base64 (a base64 FileDescriptorSet from protoc --descriptor_set_out --include_imports) instead of schema_file. The decoded message is exposed to jq_filter as payload_decoded (field names as in the schema, enums as their names, 64-bit ints as JSON numbers, default-valued fields included). A record that fails to decode with the given schema is treated as not matching. Like jq_filter, payload_protobuf only works with the default librdkafka client, and does not support schema-registry framing (magic byte + schema id prefix).
Notes to convey:
queue_mode is optional: steal (default, only your local app gets a matched message) or mirror (both your app and the deployed app get a copy).filter/message_filter and a jq_filter are both set, both must match.jq_filter requires operator 3.183.0+, CLI 3.232.0+, and the default librdkafka client — it is not supported with the Java client (Kafka Streams), which fails with a clear error.queue_id per entry — it also accepts payload_protobuf per entry.If the user has the mirrord-config skill, point them there for the full mirrord.json.
MirrordPropertyList (in the target's namespace, or the operator's namespace if sharing) has bootstrap.servers; does not set group.id.MirrordSplitConfig is in the target's namespace with spec.targetRef (apiVersion, kind, name).id, kind: kafka, a clientConfig (or spec.clientConfigs.kafka), and appConfig.topic.appConfig.groupId or appConfig.appId.kind (targetRef) is one of Deployment, StatefulSet, Rollout.clientConfig resolves to a MirrordPropertyList, looked up in the target's namespace first, then the operator's namespace (operator 3.191.0+) — or, as a final legacy fallback, a MirrordKafkaClientConfig of that name in the operator namespace.target matches the MirrordSplitConfig targetRef.jq_filter is only used with librdkafka (not with mirrord.client_implementation: java).payload_protobuf is only used with librdkafka, on queue_type: Kafka, and sets exactly one of schema_file or descriptor_base64 plus message_type.min.insync.replicas / acks workaround.mirrord.ssl.*.base64); offer PEM conversion commands only for older operators.podFile source (operator 3.201.0+) can read them straight from the rendered file.mirrord-tmp-* topics.jq_filter won't work.Present results as:
✅ Validation passed
⚠️ Warning: [description + workaround]
❌ Error: [what's wrong + how to fix]
Full setup: brief overview of the 2 resources → MirrordPropertyList YAML → MirrordSplitConfig YAML → example mirrord.json → validation → warnings.
Single resource: YAML → validation → warnings.
Troubleshooting: read references/known-issues.md, use the Quick Symptom Lookup, ask for the operator version (kubectl get deploy mirrord-operator -n mirrord -o jsonpath='{.spec.template.spec.containers[0].image}'), match symptoms, suggest checking operator logs (kubectl logs -n mirrord deployment/mirrord-operator --tail 100).
"Set up Kafka splitting for my deployment" → ask for bootstrap servers, auth, workload name/namespace, topic + group-id env vars → generate MirrordPropertyList + MirrordSplitConfig + mirrord.json example.
"Filter by message body / a field in the payload" → use jq_filter (this is now supported). Confirm operator 3.183.0+/CLI 3.232.0+ and librdkafka (not Streams).
"Our topic carries raw protobuf, not JSON" → use payload_protobuf (schema_file + message_type, or descriptor_base64) alongside jq_filter on the decoded payload_decoded field. librdkafka only, same as jq_filter.
"Our topic/group name comes from a Vault-injected file, not an env var" → use a podFile source on appConfig.topic/groupId/appId (operator 3.201.0+) instead of env/volume.
"We use Kafka Streams" → appConfig.appId + mirrord.client_implementation: java + operator.kafkaSplittingSidecar.enabled: true. Note jq_filter is unavailable with the Java client.
"We use AWS MSK with IAM" → mirrord.auth.kind: MSK_IAM + mirrord.auth.aws_region; annotate the operator SA with the role ARN via sa.roleArn.
"We use JKS for Kafka auth" → put the same ssl.* properties the JVM app already uses on the MirrordPropertyList: base64-encode the store into mirrord.ssl.truststore.base64/mirrord.ssl.keystore.base64 (or point ssl.truststore.location/ssl.keystore.location at a store mounted into the operator pod), via a Secret. Requires operator 3.199.0+ — on older operators, fall back to JKS→PEM conversion and ssl.*.pem via a Secret. See references/mirrord-property-list-crd.md.
"My session fails with INCONSISTENT_GROUP_PROTOCOL" / "We use KafkaJS" → set mirrord.temporary_group_id: "true" on the MirrordPropertyList (operator 3.195.0+). The operator then patches the consumer group to a generated temporary one so it never negotiates a protocol with the app's client. Only reach for the Kafka Streams JVM-proxy setup (appConfig.appId + client_implementation: java) if the workload is an actual Kafka Streams app.
"Splitting fails with a PolicyViolation broker error" / "We use Confluent Cloud" → the managed platform enforces a minimum replication factor for new topics. Set mirrord.split_topic.replication_factor: copy (or a number matching the platform's minimum) on the MirrordPropertyList (operator 3.191.0+).
"My session times out" → check known-issues (single-replica min.insync.replicas, ephemeral topic cleanup), tune spec.restart.timeout, check operator logs.
"Migrate our existing Kafka splitting config" → map MirrordKafkaTopicsConsumer→MirrordSplitConfig and MirrordKafkaClientConfig→MirrordPropertyList (mapping tables in the reference files). You can migrate the topics consumer first — clientConfig falls back to the legacy client config by name.
MirrordKafkaTopicsConsumer/MirrordKafkaClientConfig for a new setup — use MirrordSplitConfig + MirrordPropertyList.group.id — mirrord manages it.MirrordPropertyList to the operator's namespace — the target's namespace is still the recommended default; only use the operator's namespace (operator 3.191.0+) when the user wants to share one connection config across namespaces.appConfig.groupId and appConfig.appId on one queue.jq_filter for Kafka Streams (Java client) sessions — it's librdkafka-only.jq_filter supports it.