transform_messages
Rewrite Kafka messages with a JavaScript transform and append the results as new messages to repair malformed payloads, with preview and confirmation before writes.
Instructions
Rewrite messages with a JavaScript transform and write the results as new messages: to the same topic, another topic, or a topic on another configured cluster. Use it to repair messages produced in the wrong shape, for example a document wrapped in a payload field: transform "return { value: value.payload }".
The transform is JavaScript run once per message. It sees the same variables as search_messages and returns an object with any of value, key and headers; what it leaves out is kept from the original, and null skips the message. An integer beyond 2^53 arrives as a BigInt so ids are written back exactly. An optional script selects which messages to transform. Schema-encoded values (Avro, Protobuf, JSON Schema) are decoded for the transform and re-encoded with the same schema id; set value_schema to encode to a different schema or registry.
The originals are never changed or deleted: Kafka cannot edit a message, so the new messages are appended. Writing back to the source topic stops at the end offset the call started with, so it never reads its own output. Every new message carries provenance headers naming the source topic, partition and offset.
Nothing is written to Kafka unless confirm is true. Without it the response counts what would be written, shows before/after examples and groups failures by reason with example addresses. download saves the new messages on the user's machine instead, as produce_message items, one per line, so the set can be inspected and later sent with produce_message upload: over stdio at once, over HTTP through a link and command the transform runs behind when fetched. Nothing is stored on the server, and downloading works on a read-only cluster.
Partitions are processed in ascending order and writing is not atomic: messages written before a timeout or failure stay. complete, stopped_reason and scanned_ranges say how far the run got; resume with partitions and from_offset.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| topic | Yes | Topic to read the messages from. Matched exactly and case-sensitively. | |
| script | No | Optional JavaScript predicate, as in search_messages, choosing which messages to transform. Return true to transform the message. Omit to transform every message in the range. | |
| confirm | No | Optional. When false or omitted, nothing is written to Kafka: the response counts what would be written, shows examples and groups the failures. Must be true to produce the new messages. | |
| download | No | Optional. Instead of writing to Kafka, save every new message to a file on the user's own machine, one produce_message item per line, so the set can be inspected and later sent with produce_message upload. Over stdio the file is saved right away and download.path names it; over HTTP the response is a link and a command, e.g. curl, to run on the user's machine, and the transform runs when it is fetched. Nothing is stored on the server and nothing is written to Kafka, so it works on a read-only cluster; it cannot be combined with confirm. | |
| to_offset | No | Optional exclusive offset to stop at, applied to every partition read. | |
| transform | Yes | JavaScript that returns the new message, with the same variables as search_messages' script: value (parsed JSON or the decoded Avro/Protobuf/JSON Schema record; text otherwise), key, headers, partition, offset, timestamp, value_bytes, key_bytes, format, schema_id and decode_error. Return an object with any of value, key and headers; a field left out is kept from the original. Return null to skip the message. Example: return { value: value.payload }. A string value is written as its text, any other value as JSON, and a schema-encoded or topic_formats topic is re-encoded to its schema. Returning headers replaces all of them; spread the originals to keep them and set one to null to remove it: { headers: { ...headers, 'error-reason': null } }. Standard JavaScript (ES2020: arrow functions, spread, destructuring, regular expressions, JSON, Math, Date) is available; console, require, timers and network are not. An integer beyond 2^53 arrives as a BigInt so it is written back exactly: pass it through, compare it with a BigInt literal such as 123n, or use String(); turning it into a Number rounds it, and the message is refused rather than written. | |
| key_schema | No | Optional. Like value_schema, for the key. | |
| partitions | No | Optional partitions to read. Defaults to every partition. Partitions are processed one after another in ascending order. | |
| from_offset | No | Optional inclusive offset to start at, applied to every partition read. Use it with partitions to resume a run that stopped. | |
| max_messages | No | Optional maximum number of messages to transform, counting only those the script selects. Defaults to 100000, at most 10000000. complete is false when the limit stops the run. | |
| to_timestamp | No | Optional exclusive end time (RFC3339). Messages whose own timestamp is at or after it are skipped and not counted as scanned. | |
| value_schema | No | Optional. Encode every new value, as JSON, to this Schema Registry schema on the destination cluster. Needed when a schema-encoded value moves to a cluster with a different registry, or to change the schema. An empty object {} uses the latest version of <destination_topic>-value. Without it a value read with a schema id is written with that same id. | |
| from_timestamp | No | Optional inclusive start time (RFC3339). Messages whose own timestamp is earlier are skipped and not counted as scanned. | |
| keep_partition | No | Optional. Write each message to the partition number it was read from, instead of letting its key decide. The destination needs at least as many partitions. Use it when the source producer chose partitions explicitly. | |
| max_value_bytes | No | Optional maximum number of value bytes to show in each example. Defaults to 4096. Affects the response only. | |
| timeout_seconds | No | Optional wall-clock limit in seconds. Defaults to 300, at most 3600. Messages already written when it fires stay written; scanned_ranges says how far each partition got. | |
| destination_topic | No | Optional topic to write the new messages to. Defaults to the source topic, where the new messages are appended after the originals; the originals stay. It must already exist. | |
| destination_cluster | No | Optional cluster to write to. Defaults to the cluster this endpoint serves. Use list_clusters to see which names are valid. It must be writable when confirm is true. |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| note | No | ||
| topic | Yes | ||
| errors | Yes | ||
| failed | Yes | ||
| applied | Yes | ||
| matched | Yes | ||
| scanned | Yes | ||
| skipped | Yes | ||
| written | Yes | ||
| complete | Yes | ||
| download | No | ||
| examples | Yes | ||
| warnings | No | ||
| transformed | Yes | ||
| file_written | No | ||
| scanned_ranges | Yes | ||
| source_cluster | Yes | ||
| stopped_reason | Yes | ||
| errors_truncated | No | ||
| destination_topic | Yes | ||
| destination_cluster | Yes |