kafka
GitHub管理Kafka集群、主题、消费者组、ACL、Schema及连接器等资产。提供基于exec命令的CRUD操作,支持配置变更、权限管理及数据重置,适用于消息队列运维与开发调试。
Trigger Scenarios
Install
npx skills add opskat/opskat --skill kafka -g -y
SKILL.md
Frontmatter
{
"name": "kafka",
"description": "Manage Kafka topics, consumer groups, ACLs, schemas, connectors and messages via exec, using a family + verb + target command syntax."
}
Kafka assets
Command syntax
<family> <verb> [target] [--flags] — family is the resource kind, verb is the
operation, and target names the topic / consumer group / subject / connector.
Every example below is literally executable as written. Optional flags are not marked with brackets; each useful combination is spelled out on its own line.
cluster
cluster overviewcluster brokerscluster broker-config --broker-id=1cluster cluster-configs
topic
topic listtopic list --search=orders --include-internal --page=1 --page-size=50topic describe orderstopic create orders --partitions=3 --replication-factor=2topic create orders --partitions=3 --replication-factor=2 --configs='{"retention.ms":"604800000"}'topic update-config orders --config-updates='[{"name":"retention.ms","value":"1000"}]'topic update-config orders --config-updates='[{"name":"retention.ms","op":"delete"}]'topic increase-partitions orders --partition-count=6topic delete-records orders --records='[{"partition":0,"offset":100}]'topic delete orders
consumer-group
consumer-group listconsumer-group describe mygroupconsumer-group reset-offset mygroup --topic=orders --mode=earliestconsumer-group reset-offset mygroup --topic=orders --mode=offset --offset=1000 --partitions='[0,1]'consumer-group reset-offset mygroup --topic=orders --mode=timestamp --timestamp-millis=1700000000000consumer-group delete mygroup
acl
acl listacl list --resource-type=topic --resource-name=orders --principal=User:aliceacl create --resource-type=topic --resource-name=orders --principal=User:alice --acl-operation=read --permission=allowacl create --resource-type=topic --resource-name=orders --principal=User:alice --acl-operation=read --permission=allow --pattern-type=prefixed --host=10.0.0.7acl delete --resource-type=topic --resource-name=orders --principal=User:alice --acl-operation=read --permission=allow --host='*'
schema
schema list-subjectsschema list-versions mysubjectschema describe mysubjectschema describe mysubject --version=3schema check-compatibility mysubject --schema='{"type":"record","name":"X","fields":[]}'schema register mysubject --schema='{"type":"record","name":"X","fields":[]}' --schema-type=AVROschema delete mysubject --version=3schema delete mysubjectschema delete mysubject --permanent
connect
connect list-clustersconnect list-connectorsconnect list-connectors --cluster=mainconnect describe myconnectorconnect create myconnector --config='{"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector","tasks.max":"1"}'connect update-config myconnector --config='{"tasks.max":"2"}'connect pause myconnectorconnect resume myconnectorconnect restart myconnectorconnect restart myconnector --include-tasks --only-failedconnect delete myconnector
message
message browse ordersmessage browse orders --partition=0 --start-mode=newest --limit=100 --decode-mode=jsonmessage browse orders --start-mode=offset --offset=1000 --limit=20message browse orders --start-mode=timestamp --timestamp-millis=1700000000000 --limit=20message inspect orders --partition=0 --offset=1000message produce orders --value='{"a":1}'message produce orders --key=k1 --value='{"a":1}' --value-encoding=text --headers='[{"key":"trace-id","value":"abc","encoding":"text"}]'
Flag reference
Every flag each verb accepts — anything else is rejected rather than ignored. Bold flags are required: omitting one does not default it, the command is rejected before it runs. Verbs not listed here take no flags at all.
cluster broker-config:--broker-idtopic list:--include-internal,--search,--page,--page-sizetopic create:--partitions,--replication-factor,--configstopic update-config:--config-updatestopic increase-partitions:--partition-counttopic delete-records:--recordsconsumer-group reset-offset:--topic,--mode,--offset,--timestamp-millis,--partitionsacl list:--resource-type,--resource-name,--pattern-type,--principal,--host,--acl-operation,--permission,--page,--page-sizeacl create:--resource-type,--principal,--acl-operation,--permission,--resource-name,--pattern-type,--hostacl delete:--resource-type,--principal,--acl-operation,--permission,--host,--resource-name,--pattern-typeschema describe:--versionschema check-compatibility:--schema,--version,--schema-type,--referencesschema register:--schema,--schema-type,--referencesschema delete:--version,--permanentconnect list-connectors:--clusterconnect describe:--clusterconnect create:--config,--clusterconnect update-config:--config,--clusterconnect pause:--clusterconnect resume:--clusterconnect restart:--cluster,--include-tasks,--only-failedconnect delete:--clustermessage browse:--partition,--start-mode,--offset,--timestamp-millis,--limit,--max-bytes,--decode-mode,--max-wait-millismessage inspect:--partition,--offset,--max-bytes,--decode-mode,--max-wait-millismessage produce:--partition,--key,--key-encoding,--value,--value-encoding,--headers,--timestamp-millis
Flag values
Checked before any approval dialog, so a wrong value costs nothing:
- Numeric flags (
--limit,--offset,--page,--partitions, …) must be plain decimal integers:1000, not1,000,1_000,1e3or3.0. - Boolean flags (
--include-internal,--permanent,--include-tasks,--only-failed) may be written bare to mean true (--permanent) or with an explicit--permanent=true/--permanent=false. No other value is accepted.
Checked only once the command runs — i.e. after you have been approved, so a wrong value here means an approved command that then fails:
consumer-group reset-offset --mode:earliest,latest,offset,timestamp(defaultlatest).timestampmode needs--timestamp-millisor it fails.offsetmode does not check for--offset: omitting it is read as--offset=0, which resets every selected partition to the start of the topic with no error. Always pass--offsetexplicitly in that mode.message browse --start-mode:newest,oldest,offset,timestamp(defaultnewest). Note it isnewest/oldesthere, notlatest/earliest— those two spellings belong toreset-offset --mode.--decode-mode(browse/inspect):text,json,hex,base64(defaulttext).--key-encoding/--value-encoding(produce):text,json,hex,base64(defaulttext);hex/base64decode the value before producing it.acl --permission:allowordeny— plusanyonacl listonly, which is also its default there.acl --acl-operation:read,write,create,delete,alter,describe,cluster_action,describe_configs,alter_configs,idempotent_write,create_tokens,describe_tokens,all— plusanyonacl listonly, which is also its default there.acl --resource-type:topic,group,cluster,transactional_id,delegation_token— plusanyonacl listonly, which is also its default there. (useris parsed but not supported.)acl --pattern-type: onacl create/acl deleteonlyliteral(default) orprefixed. Onacl listall four ofany,match,literal,prefixedare accepted, and the default is notliteral: it isany, ormatchwhen--resource-nameis given. Separators and case are ignored throughout, sotransactional_id,transactionalIdandTRANSACTIONAL-IDare the same value.
Not checked on this side at all — the broker, Schema Registry or Connect cluster is the only thing that will reject a wrong value, after approval and a round trip:
--schema-type(schema register / check-compatibility):AVRO(the registry's default when omitted),JSON,PROTOBUF. Upper case, spelled exactly — the Schema Registry rejectsavro.--version(schema describe / check-compatibility / delete): a version number orlatest. Anything else is a 404 from the registry.--configs/--config-updatesnames are Kafka topic config keys,--configkeys are Kafka Connect connector properties, and--principalmust be in the broker'sUser:nameform. All are validated by the cluster, not here.
JSON-valued flags — unlisted fields are silently dropped by the JSON decoder, so spell them exactly:
--configs,--config(connect): a flat JSON object whose values are strings:'{"retention.ms":"604800000"}', not'{"retention.ms":604800000}'.--config-updates:[{"name":…,"value":…,"op":…}]. The field isname, notkey.opisset(default),delete/unset,appendorsubtract;valueis ignored for a delete.--records(delete-records):[{"partition":0,"offset":100}].--headers(produce):[{"key":…,"value":…,"encoding":…}]— here the field really iskey, unlike--config-updates.encodingistext(default),json,hexorbase64and decodes that header's value.--references(schema):[{"name":"a.avsc","subject":"a","version":1}].--partitions(reset-offset): a JSON array of partition numbers ('[0,1]'); omitting it resets every partition of the topic.
Notes
- The
schemaandconnectfamilies require Schema Registry / Kafka Connect to be configured on the asset. Neither is settable throughput_asset, so on an asset you created that way everyschema */connect *command fails with "schema registry 未启用" / "kafka connect 未启用" — configure them in the desktop UI first. - Flag names use hyphens (
--replication-factor); underscores are accepted as the same flag, but do not pass both spellings in one command. - Unknown flags are rejected rather than ignored, so a typo such as
--paritions=3fails instead of silently creating a 0-partition topic. - Single-quote any value containing JSON, spaces, braces or
*. The command line is shell-tokenized but never shell-executed:$,|,>,&produce an error rather than expanding. - Exactly one target is allowed, and only for verbs that take one:
topic list ordersis an error, not "describe orders". - A target containing whitespace cannot be used at all — permission rules match two space-separated tokens, so such a name could never be authorized.
aclcommands take no target; the resource is named by--resource-name. Onacl create/acl deleteit is required unless--resource-type=cluster; onacl listit is an optional filter — omitting it lists every ACL.acl createwithout--hostapplies to every host.- Omitting
--versiononschema deletedeletes every version of the subject, not just the latest — it is a whole-subject delete, and with--permanentit is unrecoverable. Pass an explicit--versionwhenever you mean one version. Note that both spellings ask for the same approval (schema.delete <subject>), so the approval dialog will not warn you about the difference. cluster broker-configwithout--broker-iddescribes broker0; take the id fromcluster brokersfirst.connectcommands need--clusteronly when the asset has more than one Kafka Connect cluster configured; with a single cluster it is inferred.- Approval is granted at
<action> <resource>granularity, e.g.topic.delete orders,message.write orders,consumer_group.offset.write mygroup. Cluster-wide and ACL actions carry*as the resource (acl.write *), so approving one ACL change approves both granting and revoking. - The
scopeparameter is not used by Kafka assets; the target position names the resource.
Asset config (for put_asset)
Either brokers, or host + port, must be provided — at least one path to the cluster
is required. Once sasl_mechanism is anything other than "none", username becomes
required and either password or credential_id becomes required — creation is rejected
when the username or both password sources are missing.
| field | type | required | notes |
|---|---|---|---|
brokers |
string or string[] | yes* | Comma/semicolon/newline separated host:port list, e.g. "kafka-0:9092,kafka-1:9092" |
host |
string | no | Single-broker fallback used only when brokers is omitted |
port |
number | no | Used with host when brokers is omitted; no default — pass 9092 explicitly |
username |
string | no* | Required when sasl_mechanism is not "none"; unused otherwise |
password |
string | no* | Stored encrypted; required when sasl_mechanism is not "none", unless credential_id already supplies it |
credential_id |
number | no* | Managed password credential; an alternative to inline password; 0 means none |
client_id |
string | no | Defaults to "opskat" |
sasl_mechanism |
string | no | "none" (default), "plain", "scram-sha-256", "scram-sha-512" |
tls |
boolean | no | true to enable TLS |
tls_insecure |
boolean | no | true to skip TLS certificate verification |
tls_server_name |
string | no | TLS SNI / server name override |
tls_ca_file |
string | no | Path to a CA certificate file |
tls_cert_file |
string | no | Path to a client certificate file (mTLS) |
tls_key_file |
string | no | Path to a client key file (mTLS) |
request_timeout_seconds |
number | no | Per-request timeout override |
message_preview_bytes |
number | no | Default byte budget per message when --max-bytes is omitted; --max-bytes overrides it, hard cap 1 MiB |
message_fetch_limit |
number | no | Default message count when --limit is omitted; --limit overrides it, hard cap 1000 |
ssh_asset_id |
number | no | SSH asset to tunnel through; 0 detaches |
Version History
-
d6c7a6e
Current 2026-08-16 07:28
修正kafka reset-offset模式对--offset参数的校验描述偏差,以及acl list中--resource-name可选性的说明错误。
- aeb4bc3 2026-07-24 16:28


