# Kafka write task

> Send the samples of Synnax channels to a Kafka topic as JSON records.

A Kafka write task sends each sample of its channels to one topic as a JSON record. Use it to give other systems on Kafka the data that Synnax records. Before you start, [connect a Kafka cluster](https://docs.synnaxlabs.com/reference/hardware/kafka/connect-cluster).

The task reads channels and does not command hardware, so it takes no [control authority](https://docs.synnaxlabs.com/reference/control/authority). It has no data saving option, because it records nothing in Synnax.

## Create a Kafka write task

**Console**

1. In the devices toolbar, right-click the cluster and select **Create write task**. You can also run the “Create Kafka write task” command.
2. Enter the **Topic**, such as `synnax-samples`.
3. Click **Add channel**, then select a channel in **Channel**.
4. Set the **JSON type** of the value: **Number**, **String**, or **Boolean**.
5. Click (**Start**).

The task starts to send records at once. Each record holds the value, the channel name, and the timestamp, as in [record shape](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#record-shape).

[](https://synnax.nyc3.cdn.digitaloceanspaces.com/docs/hardware/kafka/write-task/332ee9404415-dark.mp4#t=0.001)

**Python**

```python
from synnax import kafka

cluster = client.devices.retrieve(name="Plant cluster")
temperature = client.channels.retrieve("oven_temperature")

task = client.tasks.create(
    name="Oven to Kafka",
    type="kafka_write",
    rack=cluster.rack,
    config=kafka.WriteConfig(
        device=cluster.key,
        topic="synnax-samples",
        record=kafka.Record(
            channel_pointer="/channel", timestamp_pointer="/timestamp"
        ),
        channels=[kafka.WriteChannel(channel=temperature.key)],
    ),
)
task.start()
```

Pass `rack=cluster.rack`. With no rack, Python picks an embedded rack, which can be the rack of the Driver instead of the rack of the Core.

**TypeScript**

```typescript
import { kafka } from "@synnaxlabs/client";

const [cluster] = await client.devices.retrieve({
  where: { name: { eq: "Plant cluster" } },
});
const temperature = await client.channels.retrieve("oven_temperature");

const task = await client.tasks.create({
  name: "Oven to Kafka",
  rack: cluster.rack,
  type: "kafka_write",
  config: kafka.writeConfigZ.parse({
    device: cluster.key,
    topic: "synnax-samples",
    record: { channelPointer: "/channel", timestampPointer: "/timestamp" },
    channels: [{ channel: temperature.key }],
  }),
});
await task.start();
```

## Record shape

Each record is one JSON object. Three pointers set where each part goes:

* **Value pointer** -> the sample value. `/value` by default.
* **Channel pointer** -> the name of the channel. The Console sets `/channel`. Clear it to leave the name out.
* **Timestamp pointer** -> the timestamp of the sample, in the **Time format**. The Console sets `/timestamp`. Clear it to leave the timestamp out.

With the Console defaults, a sample of `oven_temperature` gives this record value:

```json
{ "value": 182.5, "channel": "oven_temperature", "timestamp": 1767225600000000000 }
```

A pointer can go deeper than one level, such as `/data/value`. The record also has a Kafka timestamp, which is the timestamp of the sample.

The timestamp of a sample comes from the index channel of its channel. A virtual channel has no index, so its samples take the time at which the Core receives them.

**Record key** sets the Kafka key of each record. **Channel name** puts each channel on its own key, so Kafka keeps the samples of one channel in order. **None** sends records with no key.

In code, `record.fields` adds the same fields to every record. A static field adds a fixed value, and a generated field adds a new UUID or the current time. The Console does not show them.

```python
from synnax import http, kafka

kafka.Record(
    channel_pointer="/channel",
    fields=[
        http.StaticWriteField(pointer="/site", json_type="string", value="plant-1"),
        http.GeneratedWriteField(pointer="/id", generator="uuid"),
    ],
)
```

The field types come from `synnax.http`. A generated timestamp field takes a `time_format`, and its default is `iso8601`.

## Task options

| Setting                                      | Default      | Notes                                                           |
| -------------------------------------------- | ------------ | --------------------------------------------------------------- |
| Topic `topic`                                |              | One topic for each task.                                        |
| Record key `record_key`                      | Channel name | **Channel name** or **None**.                                   |
| Value pointer `record.value_pointer`         | `/value`     | Empty uses `/value`.                                            |
| Channel pointer `record.channel_pointer`     | `/channel`   | Empty leaves the name out. In code, leave it unset: `""` fails. |
| Timestamp pointer `record.timestamp_pointer` | `/timestamp` | Empty leaves it out. In code, leave it unset: `""` fails.       |
| Time format `record.time_format`             | Unix (ns)    | ISO 8601, or Unix time in s, ms, µs, or ns.                     |
| Auto start `auto_start`                      | Off          | Starts the task each time the Core starts.                      |

TypeScript uses the same fields in camel case, such as `recordKey`.

## Channel reference

| Setting                    | Default | Notes                                                       |
| -------------------------- | ------- | ----------------------------------------------------------- |
| Channel `channel`          |         | The channel whose samples the task sends.                   |
| JSON type `json_type`      | Number  | **Number**, **String**, or **Boolean**.                     |
| Time format `time_format`  | None    | Required for a `timestamp` channel. Replaces the JSON type. |
| Enum mapping `enum_values` | None    | With **String**, maps numbers to text labels.               |

A string, JSON, or UUID channel always goes out as a string.

## When the cluster is unreachable

The task does not start when the Core cannot reach the cluster. While the task runs, a record that the cluster does not accept shows [`Failed to produce record`](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#failed-to-produce-record), and the task keeps running.

## Troubleshooting

* [Failed to produce record](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#failed-to-produce-record)

  * Cause

    The task could not send a record. The description gives the reason:

    * **Produce failed** -> the cluster did not accept the record, for example because the user cannot write to the topic.
    * **Channel … arrived without its index** -> a sample came with no timestamp from its index channel.
    * **Record build failed** -> a value does not fit the JSON type or the enum mapping.

    The task keeps running.

  * Fix

    * **Produce failed** -> check the topic name and the ACLs of the user.
    * **Channel … arrived without its index** -> write the index channel in the same frame as the channel.
    * **Record build failed** -> match the JSON type and the enum mapping to the values.

* [connecting to cluster: ...](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#connecting-to-cluster)

  * Cause

    The task did not start, because the Core cannot reach the cluster.

  * Fix

    Check the [cluster status](https://docs.synnaxlabs.com/reference/hardware/kafka/connect-cluster#cluster-status), then start the task again.

* [channels: at least one must be enabled](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#channels-at-least-one-must-be-enabled)

  * Cause

    The task has no enabled channels.

  * Fix

    Add a channel, or enable one.

* [channel ...: time_format is required for timestamp channel ...](https://docs.synnaxlabs.com/reference/hardware/kafka/write-task#channel-time-format-is-required-for-timestamp-channel)

  * Cause

    A `timestamp` channel has no time format.

  * Fix

    Set **Time format** on the channel.

For problems with the connection, see [Connect a Kafka cluster](https://docs.synnaxlabs.com/reference/hardware/kafka/connect-cluster#troubleshooting).