kafka-admin

kafka-admin

When to use this skill

• User mentions Kafka, Kafka topics, Kafka consumer groups.
• User mentions Schema Registry, schema evolution, Avro/Protobuf/JSON schemas in a Kafka context.
• User wants to create/manage a Kafka Connect connector.
• User wants to produce/consume Kafka messages via the CLI.

Execution base

Prefer StreamNative Cloud remote MCP when available: use fixed Kafka cluster endpoint or organization-level sncloud_context_use_cluster; after successful org-level selection, output MCP_TOOLS_REFRESH_REQUIRED instance=<instanceName> cluster=<clusterName> and stop so the runner/runtime can refresh MCP tools; after resume, use feature-gated Kafka tools (kafka_admin_*_read, kafka_admin_*_write, kafka_client_*). Remote MCP is public preview and disabled by default; cluster-level MCP must be enabled in StreamNative Cloud UI before fixed endpoints or sncloud_context_use_cluster can target a cluster. See ../cloud-core/references/remote-mcp.md. Use snctl fallback when MCP tools are unavailable or for CLI-specific examples below.

Prerequisites check

Remote MCP route: verify sncloud_context_whoami; for org-level session select cluster via sncloud_context_available_clusters / sncloud_context_use_cluster. After a successful sncloud_context_use_cluster, output exactly MCP_TOOLS_REFRESH_REQUIRED instance=<instanceName> cluster=<clusterName> and stop before using Kafka tools; ToolSearch is not a refresh substitute. If sncloud_context_use_cluster reports cluster MCP disabled/not enabled, ask user to enable cluster-level MCP in StreamNative Cloud UI or use snctl. snctl route: standard checks plus:

• The active cluster must have Kafka protocol handler (KoP) enabled. Check with snctl describe pulsarcluster <name> -n <org> — look for Kafka-related status fields/annotations.
• The acting principal needs Kafka ACLs equivalent to the requested operation (inherited from the SA's Pulsar-level permissions — see identity).

Mental model

On StreamNative Cloud, Kafka is implemented by KoP (Kafka-on-Pulsar). Kafka primitives map 1:1 to Pulsar primitives:

Kafka concept	Pulsar implementation
Topic	Persistent topic in a dedicated kafka/ namespace
Consumer group	Pulsar subscription (Failover type)
Schema Registry	Per-topic Pulsar schema storage, exposed via the KoP Schema Registry REST API
Kafka Connect	Connector runtime hosted alongside the Kafka protocol

Because of this mapping, Kafka ACLs are enforced via the underlying Pulsar role-based model.

Primary workflows

Remote MCP equivalents use documented feature-gated split tool names when enabled: kafka_admin_topics_read/write, kafka_admin_groups_read/write, kafka_admin_sr_read/write, kafka_admin_connect_read/write, kafka_admin_partitions_write, plus kafka_client_produce and kafka_client_consume. Use _read tools for list/get/status/describe and _write tools for create/update/delete/reset/compatibility/partition operations. Do not invent sncloud_resources_*, old unsplit admin names, or undeclared admin tool names.

W1. Create and list Kafka topics

snctl kafka admin topics create my-topic --partitions 3 --replication-factor 3
snctl kafka admin topics list

W2. Describe a topic

snctl kafka admin topics describe my-topic

W3. Consumer groups

snctl kafka admin groups list
snctl kafka admin groups describe my-group
# Reset group offsets. Without --execute, reset-offsets is a dry run.
snctl kafka admin groups reset-offsets my-group \
  --topic my-topic --oldest --execute
snctl kafka admin groups reset-offsets my-group \
  --topic my-topic --newest --execute
snctl kafka admin groups reset-offsets my-group \
  --topic my-topic --offset 1000 --execute
snctl kafka admin groups delete my-group

W4. Schema Registry

snctl kafka admin schemas create my-topic-value \
  -f ./schema.avsc -t AVRO
snctl kafka admin schemas get my-topic-value
snctl kafka admin schemas list
snctl kafka admin schemas delete subject my-topic-value -f

W5. Kafka Connect connectors

snctl kafka connect list
snctl kafka connect create --name my-connector --config ./connector.json
snctl kafka connect status my-connector
snctl kafka connect restart my-connector
snctl kafka connect pause my-connector
snctl kafka connect resume my-connector
snctl kafka connect delete my-connector

W6. Produce and consume (CLI helper)

snctl kafka client produce --topic my-topic --message "hello" --num-times 10
snctl kafka client consume --topic my-topic --group cli --num-messages 10 --timeout 60s

W7. Bootstrap-server discovery (for external Kafka clients)

# The bootstrap server URL is on the cluster status
snctl describe pulsarcluster <cluster-name> -n <org> -o yaml | grep -A1 kafka

Reference index

• references/topics-groups.md — topic lifecycle, consumer group semantics, offset management.
• references/partitions.md — partition count limits, reassignment.
• references/schemas.md — AVRO / JSON / Protobuf, compatibility modes.
• references/kafka-connect.md — connector config shape, supported plugins.
• references/produce-consume.md — key/value formats, headers, idempotence.

Pitfalls and gotchas

• Remote MCP Kafka admin/client tools are feature-gated; if a tool is absent, use fixed endpoint with needed feature or fall back to snctl.
• sncloud_context_use_cluster works only when cluster-level MCP is enabled on the target cluster; new clusters default to disabled until enabled in StreamNative Cloud UI. After it succeeds, emit MCP_TOOLS_REFRESH_REQUIRED instance=<instanceName> cluster=<clusterName> and wait for runner/runtime refresh because selected cluster context changes the session-scoped tool surface. Do not use ToolSearch as refresh.
• --replication-factor on KoP is accepted but physical replication is governed by the underlying Pulsar cluster, not the Kafka config. Setting replication-factor: 3 on a cluster with bookkeeper.replicas: 1 is meaningless.
• Resetting consumer-group offsets affects ALL consumers in the group; coordinate with the user.
• Schema compatibility defaults to BACKWARD; switching to NONE disables compatibility checks and allows breaking changes.
• snctl kafka connect connector configs are JSON (not YAML) — mirror the upstream Kafka Connect REST API config shape.
• Produce/consume via snctl is for smoke-testing only; real workloads should use the native Kafka client pointing at the KoP bootstrap server.
• For smoke tests, start grouped consumers before producing live messages. Historical --from-beginning reads can hang or leave the group briefly Stable after forced process stops on KoP-backed clusters.
• Partition count cannot be decreased. Increasing partitions on a topic that relies on keyed ordering breaks ordering guarantees for existing keys.