This document summarizes the Cursus TCP command API. The canonical response contract is defined in Protocol Specification; this page is a command-oriented quick reference.
Cursus accepts only Wire v2: a required binary handshake followed by correlated CRS2 frames with a 32-byte header, CRC32C, explicit compression, command ID, status, and request ID. Most application requests carry a CRQ2 command payload; publish batches use CBV2. CONSUME, STREAM, and READ_STREAM can return multiple correlated stream frames. See the Protocol Specification for the canonical byte layout.
Control-plane commands use one of these response forms:
OK
OK key=value [key=value ...]
{"status":"OK", ...}
ERROR: <code> [key=value ...]
Clients should treat OK, OK ..., and JSON responses with status:"OK" as success. Clients should treat every ERROR: response as failure and branch on the machine-readable error code immediately after the prefix.
Wire v2 errors are always encoded as typed binary payloads with code, class, retryability, message, and fields. The text forms below describe the handler-level semantics, not an alternate transport.
The first frame is the Wire v2 binary negotiation request. It selects exactly protocol version 2 and one mutually supported compression. PROTOCOL_INFO and application-level NEGOTIATE are not commands. Every data, coordinator, partition, admin, and broker-internal connection performs its own handshake.
CREATE topic=<name> [partitions=<N>] [idempotent=<bool>] [event_sourcing=<bool>] [replication_factor=<N>] [min_in_sync_replicas=<N>] [cleanup_policy=<delete|compact|delete,compact>] [retention_hours=<N>] [retention_bytes=<N>] [partitioner=<hash_key|round_robin>] [auth_policy=<open|deny_write|deny_read|acl>] [read_acl=<principal[,principal]>] [write_acl=<principal[,principal]>]
Creates a topic or patches an existing definition. Defaults apply only when the topic is missing. On an existing topic, omitted fields are preserved and explicit 0, false, or empty ACL values are authoritative. Partitions can only increase; replication factor, idempotent mode, and event-sourcing mode are immutable after creation.
Success:
OK topic=<name> partitions=<N> cleanup_policy=<policy> partitioner=<policy> auth_policy=<policy> read_acl=<csv> write_acl=<csv> retention_hours=<N> retention_bytes=<N> revision=<N> replication_factor=<N> idempotent=<bool> event_sourcing=<bool> lifecycle_epoch=<N> min_in_sync_replicas=<N|default> effective_min_in_sync_replicas=<N>
Clients decode response fields by name and must not depend on presentation order.
Common errors:
ERROR: missing_topic expected="CREATE topic=<name> [partitions=<N>]"
ERROR: invalid_topic_name topic=<name> reason="..."
ERROR: invalid_partitions reason="must be a positive integer"
ERROR: invalid_replication_factor reason="must be a positive integer"
ERROR: invalid_min_in_sync_replicas value=<N> replication_factor=<N>
ERROR: invalid_topic_policy field=cleanup_policy reason="..."
ERROR: unsupported_topic_policy field=cleanup_policy reason="..."
ERROR: create_topic_failed reason="..."
DELETE topic=<name> [if_exists=<true|false>]
Deletes a topic through the admin-only lifecycle path. With if_exists=false, a missing topic returns topic_not_found. Use if_exists=true only when an explicitly approved GitOps deletion must be safely retried. Active consumer-group members and open/committing transactions block deletion. Inactive groups and offsets, producer sequence state, target-topic operations in terminal transactions, and event-sourcing state are removed with the topic lifecycle. __consumer_offsets cannot be deleted.
Success:
OK topic=<name> deleted=true
OK topic=<name> deleted=false
OK topic=<name> deleted=true cleanup_pending=true
cleanup_pending=true means the durable logical deletion succeeded and node-local physical cleanup is queued for reconciliation; it is not permission to recreate the name until every broker has converged.
Common errors:
ERROR: missing_topic expected="DELETE topic=<name>"
ERROR: invalid_topic_name topic=<name> reason="..."
ERROR: invalid_if_exists value="..."
ERROR: topic_not_found topic=<name>
ERROR: topic_delete_blocked topic=<name> reason="..."
ERROR: internal_topic_delete_forbidden topic=<name>
ERROR: delete_topic_failed reason="..."
ALTER_TOPIC_CONFIG topic=<name> min_in_sync_replicas=<N|default>
The integer must be between 1 and the topic replication factor. default removes the durable topic override and uses the broker value. Omitting the optional field from an existing manifest or snapshot has the same fallback meaning.
Success:
OK topic=<name> min_in_sync_replicas=<N|default> effective_min_in_sync_replicas=<N>
TRUNCATE topic=<name> expected_revision=<N>
Resets the topic’s data while retaining its definition and identity. This command is admin-only and requires the last observed non-zero definition revision. A mismatch fails with topic_revision_conflict; there is no unguarded form.
Success:
OK topic=<name> truncated=true revision=<N> lifecycle_epoch=<N> leo=0 hwm=0
OK topic=<name> truncated=true revision=<N> lifecycle_epoch=<N> leo=0 hwm=0 cleanup_pending=true
Success advances both definition revision and lifecycle epoch, empties all partitions, removes inactive groups and offsets, clears producer sequence and terminal transaction references, and resets event-sourcing state. Active group members and open/committing transactions block the operation. cleanup_pending=true means the new logical lifecycle is committed but the topic remains fenced on a broker until local cleanup converges.
Common errors:
ERROR: missing_expected_revision command=TRUNCATE
ERROR: invalid_expected_revision value="..."
ERROR: topic_revision_conflict topic=<name> expected=<N> reason="..."
ERROR: topic_truncate_blocked topic=<name> reason="..."
ERROR: internal_topic_truncate_forbidden topic=<name>
ERROR: truncate_topic_failed topic=<name> reason="..."
Do not emulate truncation with DELETE plus CREATE: deletion changes topic identity and does not provide lifecycle-epoch fencing.
LIST
Success:
OK count=<N> topics=<comma-separated-topic-names>
When no topics exist, the broker returns:
OK count=0 topics=
DESCRIBE topic=<name>
Success is a JSON topic metadata object with status:"OK". Its additive definition object contains the same revision, replication factor, modes, partition count, and policy returned by CREATE; the existing partitions and policy fields remain available.
Common errors:
ERROR: missing_topic expected="DESCRIBE topic=<name>"
ERROR: topic_not_found topic=<name>
ERROR: marshal_metadata_failed reason="..."
HELP
Success:
OK commands=<comma-separated-command-names>
PUBLISH topic=<name> [partition=<N>] [key=<routing-key>] [producerId=<id>] [seqNum=<N>] [epoch=<N>] [isIdempotent=<bool>] [acks=0|1|all|-1] message=<payload>
Because message= captures the rest of the line, put optional parameters before message=.
Transaction metadata fields are not accepted on client PUBLISH; use the transaction commands for transactional records. Direct transactional_id, transaction_state, or transaction_marker injection returns ERROR: transaction_metadata_forbidden command=PUBLISH.
acks belongs to the publish request or publisher configuration; it is not topic metadata. For acks=1 or acks=all, success is a JSON ack response with status:"OK". acks=0 emits no external response frame. acks=1 responds after the leader’s durable local append while bounded ordered follower replication continues; a leader failure before commit can lose the record. acks=all and acks=-1 are aliases: the broker first requires the current ISR to meet the topic-effective minimum, then waits for the captured ISR and lifecycle-fenced committed HWM. Read-committed consumers remain bounded by committed HWM. Non-ISR failures do not delay the success response.
PUBLISH may include partition=<N> to target a partition explicitly; otherwise the topic partition policy selects the partition. Idempotent publish uses (producerId, epoch, seqNum) per partition: each new (producerId, epoch) sequence starts at seqNum=1, higher epochs fence older producer sessions, lower epochs are rejected as stale, and seqNum=0 disables dedup for that message. Distributed recovery writes and accepts only FSM snapshot version 9 with explicit committed-HWM provenance; older persistent state requires a full clean bootstrap.
The effective minimum is the topic min_in_sync_replicas override when present and broker min_insync_replicas otherwise. Standalone has one replica, so 1, all, and -1 share the local durable-append completion point when the effective minimum is 1; all/-1 reject when it is greater than 1. Idempotent publishers must use all or -1; 0 and 1 fail before append or sequence mutation.
Common errors:
ERROR: missing_topic command=PUBLISH
ERROR: missing_message command=PUBLISH
ERROR: topic_not_found topic=<name>
ERROR: partition_not_found partition=<N>
ERROR: invalid_acks value=<value>
ERROR: invalid_seq_num reason="..."
ERROR: invalid_epoch reason="..."
ERROR: stale_producer_epoch reason="..."
Internal replication command used between brokers. In distributed mode this command requires internal_token=<shared-token> before payload=. External clients and SDKs must not call it directly.
Success:
OK
Common errors:
ERROR: internal_auth_not_configured command=REPLICATE_MESSAGE
ERROR: internal_command_unauthorized command=REPLICATE_MESSAGE
ERROR: missing_payload command=REPLICATE_MESSAGE
ERROR: unmarshal_failed reason="..."
ERROR: topic_not_found topic=<name>
ERROR: partition_not_found partition=<N>
ERROR: cluster_metadata_unavailable command=REPLICATE_MESSAGE
ERROR: partition_metadata_not_found topic=<name> partition=<N>
ERROR: missing_leader_fence command=REPLICATE_MESSAGE
ERROR: NOT_PARTITION_LEADER leader=<broker-id> requested_leader=<broker-id>
ERROR: STALE_LEADER_EPOCH current=<N> requested=<N>
ERROR: invalid_commit_watermark reason="..."
ERROR: replica_append_failed reason="..."
ERROR: replica_index_prepare_failed reason="..."
ERROR: replica_index_failed reason="..."
CONSUME topic=<name> group=<group> partition=<N> offset=<N> member=<member-id> [isolation=<read_committed|read_uncommitted>] [batch=<N>]
CONSUME returns binary message frames. For consumer groups, the broker uses the committed offset for (topic, group, partition) as the authoritative resume point when one exists; otherwise the earliest offset policy is 0. CONSUME is a stateless partition-leader read: ownership, liveness, and generation fencing are enforced by coordinator commands, not on the data path. The default isolation is read_committed, which hides unresolved and aborted transactional records. isolation=read_uncommitted returns the raw committed log, including transaction metadata and control markers.
Common errors:
ERROR: invalid_consume_syntax
ERROR: missing_topic command=CONSUME
ERROR: missing_partition command=CONSUME
ERROR: missing_offset command=CONSUME
ERROR: missing_member command=CONSUME
ERROR: invalid_partition
ERROR: invalid_offset
ERROR: invalid_isolation isolation=<value>
ERROR: NOT_LEADER leader=<host:port>
ERROR: OFFSET_OUT_OF_RANGE requested=<N> earliest=<N> latest=<N>
Continuous push-mode consume command.
STREAM topic=<name> group=<group> partition=<N> offset=<N> member=<member-id> [isolation=<read_committed|read_uncommitted>]
STREAM returns one or more correlated Wire stream frames:
[CBV2 batch payload]
[empty payload] # keepalive
[STREAM_CONTROL type=CLOSE ... payload] # stream end
Clients must treat zero-length frames as keepalive. Like CONSUME, STREAM is a stateless partition-leader data path and does not validate group ownership or generation on every read. STREAM uses the same isolation contract as CONSUME; the default is read_committed. A STREAM_CONTROL type=CLOSE frame is a graceful terminator; reason=offset_out_of_range means the requested stream offset is older than the retained log. Clients should close the socket and resume through the consumer group offset contract or reset according to policy. Raw TCP disconnect without a close control frame remains possible on broker crash or network failure and should be treated as retryable.
The broker does not commit offsets when records are written to a stream socket or when the stream closes. The client must commit the next offset explicitly after processing, using its current member and generation. This keeps stream delivery consistent with the at-least-once consumer-group contract.
Common errors:
ERROR: invalid_stream_syntax
ERROR: missing_topic command=STREAM
ERROR: missing_partition command=STREAM
ERROR: missing_offset command=STREAM
ERROR: missing_member command=STREAM
ERROR: invalid_isolation isolation=<value>
ERROR: NOT_LEADER leader=<host:port>
REGISTER_GROUP topic=<name> group=<group>
REGISTER_GROUP topics=<topic-1>,<topic-2>[,...] group=<group>
REGISTER_GROUP pattern=<glob> group=<group>
Success:
OK group=<group> topic=<name> registered=true
The multi-topic forms durably register the concrete topic-partition expansion. Multi-topic members omit topic from JOIN_GROUP/SYNC_GROUP and receive topic_assignments=<topic>:P<partition>,....
JOIN_GROUP topic=<name> group=<group> member=<member-id>
A fresh join atomically registers a missing group for the topic. Use
REGISTER_GROUP only when an empty group must be provisioned before its first
member.
Success:
OK member=<assigned-member-id> generation=<N>
SYNC_GROUP topic=<name> group=<group> member=<assigned-member-id>
Success:
OK member=<assigned-member-id> generation=<N> assignments=<partition-list>
HEARTBEAT topic=<name> group=<group> member=<assigned-member-id> [generation=<N>]
Success:
OK member=<assigned-member-id> generation=<N>
LEAVE_GROUP topic=<name> group=<group> member=<assigned-member-id>
Success:
OK group=<group> member=<assigned-member-id> left=true
LIST_OFFSETS topic=<name> [partition=<N>]
Returns retained and readable offset bounds for all partitions or one partition.
OK topic=<name> partitions=<N> offsets=P0:earliest=<N>:latest=<N>:leo=<N>:hwm=<N>,P1:earliest=<N>:latest=<N>:leo=<N>:hwm=<N>
latest is the next readable committed offset and is the value SDKs should use for auto_offset_reset=latest. leo is the log end offset, and hwm is the high-water mark before the broker caps reads to the flushed durable tail.
Errors:
ERROR: missing_topic command=LIST_OFFSETS
ERROR: topic_not_found topic=<name>
ERROR: invalid_partition command=LIST_OFFSETS
ERROR: partition_not_found partition=<N>
FETCH_OFFSET topic=<name> group=<group> partition=<N>
Success:
OK offset=<nextOffset>
When no offset has been committed, the broker returns OK offset=0.
COMMIT_OFFSET topic=<name> group=<group> partition=<N> offset=<nextOffset> member=<member-id> generation=<N>
The offset is the next offset to read after successful processing. Commits are monotonic per (topic, group, partition): a commit lower than the current offset fails and does not rewind the group. member and generation are required, and the member must own the partition in that generation.
Success:
OK
Common errors:
ERROR: invalid_offset
ERROR: missing_generation command=COMMIT_OFFSET
ERROR: invalid_generation command=COMMIT_OFFSET
ERROR: offset_regression reason="..."
ERROR: GEN_MISMATCH current=<N> requested=<N> group=<group> member=<member-id>
ERROR: NOT_OWNER partition=<N> member=<member-id> group=<group> generation=<N>
ERROR: member_not_found member=<member-id> group=<group>
ERROR: offset_manager_not_available
ERROR: commit_offset_failed reason="..."
BATCH_COMMIT topic=<name> group=<group> member=<member-id> generation=<N> offsets=P0:<nextOffset>,P1:<nextOffset>
Success:
OK batched=<N>
The P prefix in each partition entry is required. The broker validates member and generation before applying the batch and rejects the whole batch if any partition is no longer owned by that member.
Common errors:
ERROR: invalid_batch_commit_format
ERROR: invalid_batch_commit_entry reason=<reason>
ERROR: missing_generation command=BATCH_COMMIT
ERROR: invalid_generation command=BATCH_COMMIT
ERROR: duplicate_partition partition=<N> group=<group> topic=<topic>
ERROR: no_valid_offsets
ERROR: offset_regression reason="..."
ERROR: GEN_MISMATCH current=<N> requested=<N> group=<group> member=<member-id>
ERROR: NOT_OWNER partition=<N> member=<member-id> group=<group> generation=<N>
ERROR: member_not_found member=<member-id> group=<group>
ERROR: offset_manager_not_available
ERROR: bulk_commit_failed reason="..."
GROUP_STATUS group=<group>
Success is a JSON group status response with status:"OK".
FIND_COORDINATOR group=<group>
Success:
OK host=<host> port=<port>
In distributed mode, non-coordinator brokers can return:
ERROR: NOT_COORDINATOR host=<host> port=<port>
METADATA topic=<name>
Success is a text response:
OK topic=<name> partitions=<N> leaders=<csv> epochs=<csv> revision=<N> replication_factor=<N> idempotent=<bool> event_sourcing=<bool> cleanup_policy=<policy> partitioner=<policy> auth_policy=<policy> read_acl=<csv> write_acl=<csv> retention_hours=<N> retention_bytes=<N> lifecycle_epoch=<N>
Common errors:
ERROR: missing_topic command=METADATA
ERROR: topic_not_found topic=<name>
ERROR: fsm_not_available
ERROR: distribution_not_enabled
CLUSTER_STATUS returns OK cluster=<json> in distributed mode. The JSON document contains the Raft leader address, active/inactive broker counts, partition leader epochs and committed HWMs, plus leaderless and under-replicated partition totals. A leader is considered available only when its registered broker is active.
Common errors are ERROR: distribution_required command=CLUSTER_STATUS, ERROR: fsm_not_available command=CLUSTER_STATUS, and ERROR: marshal_cluster_status_failed reason="...".
ELECT_LEADER topic=<name> partition=<N> broker=<broker-id> performs a controlled preferred-leader change through the Raft metadata log. The target must be an active replica already present in the partition ISR. The broker records the current leader epoch in the command and the FSM rejects stale concurrent changes. A successful change increments leader_epoch while preserving committed_hwm, replicas, and ISR. Retrying an election whose target is already leader succeeds with changed=false and does not advance the epoch again.
Success:
OK topic=<name> partition=<N> previous_leader=<broker-id> leader=<broker-id> leader_epoch=<N> changed=<true|false>
Common errors:
ERROR: distribution_required command=ELECT_LEADER
ERROR: missing_broker command=ELECT_LEADER
ERROR: partition_not_found topic=<name> partition=<N>
ERROR: leader_election_rejected topic=<name> partition=<N> broker=<id> reason="..."
ERROR: leader_election_result_unavailable topic=<name> partition=<N>
This command does not add replicas, expand ISR, or perform data movement. Reassignment and broker draining require a separate catch-up-aware workflow.
Internal replication command used by distributed brokers. In distributed mode this command requires internal_token=<shared-token> before type=. External clients and SDKs must not call it directly.
Success:
OK
Common errors:
ERROR: internal_auth_not_configured command=RAFT_APPLY
ERROR: internal_command_unauthorized command=RAFT_APPLY
ERROR: missing_required_params command=RAFT_APPLY params=type,payload
ERROR: empty_required_params command=RAFT_APPLY params=type,payload
ERROR: distribution_required command=RAFT_APPLY
ERROR: invalid_payload_json reason="..."
ERROR: raft_apply_failed reason="..."
Event-sourcing commands are routed by aggregate key. In distributed mode, the broker handling the command must be the leader for the aggregate partition. A non-leader broker returns:
ERROR: NOT_LEADER leader=<host:port>
Clients and SDKs should reconnect to that leader and retry. Followers index replicated event-sourcing records, apply quorum-replicated snapshots, and can pull missing snapshots from the partition leader with token-authenticated internal catch-up commands after restart. Partitions restore a synced high-watermark checkpoint with durable-tail clamping, so committed reads remain bounded by the last successful committed tail.
INIT_PRODUCER_ID transactional_id=<id>
Initializes or reinitializes a broker-managed producer session for a transactional id. Success: OK transactional_id=<id> producerId=<producer-id> epoch=<N>. Reinitialization bumps epoch and fences older producers for that transactional_id. If the transaction is already committing, the broker rejects reinitialization so the prepared commit can be retried or recovered. After transactional_id_expiration_ms, completed transactions discard staged payloads but retain a compact producer-epoch tombstone in coordinator snapshots. Active open and committing transactions remain available for recovery.
BEGIN_TXN transactional_id=<id> producerId=<producer-id> epoch=<N>
Starts a broker-managed transaction using the producerId and epoch returned by INIT_PRODUCER_ID. In distributed mode, transactional_id maps to a durable logical coordinator shard; route transaction commands to the owner returned by FIND_COORDINATOR transactional_id=<id>. Shard ownership changes advance a coordinator epoch that fences the previous owner. Success: OK transactional_id=<id> state=open producerId=<producer-id> epoch=<N>. One initialized epoch may begin one transaction. After commit or abort, call INIT_PRODUCER_ID before the next begin; otherwise the broker returns producer_reinitialization_required. An uncertain END_TXN may still be retried with the completed epoch.
TXN_PUBLISH transactional_id=<id> topic=<topic> [partition=<N>] producerId=<producer-id> seqNum=<N> epoch=<N> [key=<key>] message=<payload>
Appends one unresolved record immediately through the normal partition-leader path. seqNum is required and makes retries idempotent even on non-idempotent topics. The record remains invisible to read_committed until its commit marker and final coordinator decision agree. Explicit abort and transaction_timeout_ms timeout resolution append abort markers. The producer and epoch must match BEGIN_TXN; stale epochs are fenced.
SEND_OFFSETS_TO_TXN transactional_id=<id> producerId=<producer-id> epoch=<N> topic=<topic> group=<group> member=<member> generation=<N> offsets=P<partition>:<nextOffset>,P<partition>:<nextOffset>
Stages consumer offsets in the transaction. The broker validates group member, generation, lifecycle epoch, topic-partition ownership, and monotonic offsets at stage and commit time. Callers may repeat this command for multiple topics in the same (group, member, generation, registrationEpoch) session. Finalization applies the complete set atomically under one membership fence.
END_TXN transactional_id=<id> producerId=<producer-id> epoch=<N> result=<commit|abort>
Commit validates the transaction, appends staged source offsets as unresolved records in __consumer_offsets, registers those partitions as participants, persists prepare_commit, appends markers to every output and offset partition, and finally persists committed. The same final decision exposes output and offsets; committed offsets are subsequently materialized as ordinary revised snapshots for long-term recovery. Abort persists prepare_abort, appends abort markers, and persists aborted; the timeout monitor uses the same path. Recovery retries either prepared state on its current coordinator-shard owner. Ownership and timeout work are distributed across active brokers, and the marker’s durable coordinator-shard epoch rejects stale owners after reassignment independently of producer fencing. Repeated finalization is idempotent. This provides exactly-once processing for Cursus source offsets and Cursus output records, not external side effects.
TXN_STATUS transactional_id=<id>
Returns transaction state and staged operation counts.
APPEND_STREAM topic=<name> key=<aggregate-key> version=<N> [event_type=<type>] [schema_version=<N>] [metadata=<json>] message=<payload>
Success:
OK version=<N> offset=<N> partition=<N>
Common errors:
ERROR: missing_topic
ERROR: missing_key
ERROR: missing_version
ERROR: invalid_version
ERROR: missing_message
ERROR: topic_not_found topic=<name>
ERROR: event_sourcing_not_enabled topic=<name>
ERROR: version_conflict current=<N> expected=<N>
ERROR: append_stream_failed reason="..."
READ_STREAM topic=<name> key=<aggregate-key> [from_version=<N>]
Success returns a JSON envelope frame with status:"OK", followed by a binary batch frame containing committed events. Error envelopes use JSON with status:"ERROR". from_version must be a positive integer; invalid values return invalid_from_version.
STREAM_VERSION topic=<name> key=<aggregate-key>
Success:
OK version=<N>
SAVE_SNAPSHOT topic=<name> key=<aggregate-key> version=<N> message=<payload>
Success:
OK version=<N> partition=<N>
Common errors:
ERROR: snapshot_version_exceeds_stream version=<N> current=<N>
ERROR: snapshot_save_failed reason="..."
READ_SNAPSHOT topic=<name> key=<aggregate-key>
Success when a snapshot exists:
OK snapshot={"version":500,"payload":"..."}
Success when no snapshot exists:
OK snapshot=null
enable_sasl=true, protected commands require connection or inline authentication. Every user must declare at least one of admin, topic.read, topic.write, group, transaction, or *; invalid authentication configuration fails startup. Cross-boundary commands require every applicable permission. Missing authentication returns authentication_required, and missing coarse permission returns NOT_AUTHORIZED_FOR_OPERATION. Per-topic auth_policy=acl is evaluated afterward with read_acl/write_acl and returns NOT_AUTHORIZED_FOR_TOPIC on denial. This is a token contract, not a mechanism-specific SASL byte protocol; use TLS/mTLS and network controls for broker exposure.retention_hours and retention_bytes policy metadata. 0 means broker default. Reads before the earliest retained offset fail with ERROR: OFFSET_OUT_OF_RANGE requested=<N> earliest=<N> latest=<N>. SDKs should apply auto_offset_reset (earliest, latest, or error) to decide whether to reset or fail; latest should use LIST_OFFSETS latest, the next readable committed offset.partitioner=hash_key uses FNV-1a 64-bit hash modulo partition count for keyed messages and round-robin for missing keys. partitioner=round_robin ignores keys. Increasing partition count can remap future records for an existing key.Malformed frames and handler failures also use the same error prefix:
ERROR: decode_failed reason="..."
ERROR: malformed_input reason=missing_topic_or_payload
ERROR: command_failed reason="..."
ERROR: empty_command_response
ERROR: unknown_command command=<name>
ERROR: empty_command