# Kafka read task

> Consume a Kafka topic and record values from its JSON records in channels.

A Kafka read task consumes one topic and records values from its JSON records in channels. Before you start, [connect a Kafka cluster](https://docs.synnaxlabs.com/reference/hardware/kafka/connect-cluster).

## Records and fields

A record of the task is one Kafka record key. A field is one value in the JSON value of the record, found by a [JSON pointer](https://datatracker.ietf.org/doc/html/rfc6901) such as `/temperature`, and recorded in one channel.

* **Record key** -> the task reads only records with this key into the fields of the record. An empty key, shown as **Any key**, reads each record that no other record of the task claims, records with no key included.
* **Values** -> each record value must be JSON. The task does not read Avro or Protobuf.
* **Strings** -> a string channel records the text. The **Enum mapping** of a field maps text labels to numbers.

The Console has no topic browser. Type the topic, the record keys, and the pointers.

## Create a Kafka read task

**Console**

1. In the devices toolbar, right-click the cluster and select **Create read task**. You can also run the “Create Kafka read task” command.
2. Enter the **Topic**, such as `sensor-readings`.
3. Click **Add record**, then set **Record key**, or leave it empty for any key.
4. Click **Add field** on the record, then set **Pointer** and **Data type**.
5. Click (**Start**).

The Console creates a channel for each field and an index channel for each record. It names them from the cluster name, the record key, and the pointer. To choose a name, set **Channel** before you start.

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

**Python**

Python does not create channels. Create one index channel for each record first.

```python
import synnax as sy
from synnax import kafka

cluster = client.devices.retrieve(name="Plant cluster")
time = client.channels.create(
    name="oven_time", is_index=True, data_type=sy.DataType.TIMESTAMP
)
temperature = client.channels.create(
    name="oven_temperature", index=time.key, data_type=sy.DataType.FLOAT64
)

task = client.tasks.create(
    name="Oven read",
    type="kafka_read",
    rack=cluster.rack,
    config=kafka.ReadConfig(
        device=cluster.key,
        topic="sensor-readings",
        records=[
            kafka.ReadRecord(
                record_key="oven",
                fields=[
                    kafka.ReadField(
                        pointer="/temperature", 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 does not create channels. Create one index channel for each record first.

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

const [cluster] = await client.devices.retrieve({
  where: { name: { eq: "Plant cluster" } },
});
const time = await client.channels.create({
  name: "oven_time",
  isIndex: true,
  dataType: "timestamp",
});
const temperature = await client.channels.create({
  name: "oven_temperature",
  index: time.key,
  dataType: "float64",
});

const task = await client.tasks.create({
  name: "Oven read",
  rack: cluster.rack,
  type: "kafka_read",
  config: kafka.readConfigZ.parse({
    device: cluster.key,
    topic: "sensor-readings",
    records: [
      {
        recordKey: "oven",
        fields: [{ pointer: "/temperature", channel: temperature.key }],
      },
    ],
  }),
});
await task.start();
```

## Timestamps

The **Source** of a record sets the timestamps of its samples:

* **Kafka timestamp** -> the timestamp of the Kafka record, in milliseconds. The default.
* **Record value** -> a time in the JSON value, at a **Pointer**, in a **Time format**: ISO 8601, or Unix time in seconds, milliseconds, microseconds, or nanoseconds. The value goes to the index channel of the record.

Timestamps in an index channel must increase. A topic with more than one partition can deliver records out of time order, so the task skips a record that is not after the last timestamp that it wrote and shows a [skip warning](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#skipped-of-records). Kafka timestamps only count milliseconds, so a record with a Kafka timestamp in the same millisecond as the last written timestamp goes 1 ns after it.

In code, a record value timestamp is a field whose channel is the index channel, with a `time_format`. Put the key of that field in `index` on the record.

## Consumer group and start offset

The task reads the topic as a member of a Kafka consumer group. Set **Consumer group** to share offsets with other consumers. When it is empty, the group is `kafka-` followed by the key of the task.

The task commits the offset of a record after the Core stores its samples. When the task starts again, it continues from the committed offset of the group. **Start from** sets where a group with no committed offset starts:

| Start from     | Value      | Starts at                                                       |
| -------------- | ---------- | --------------------------------------------------------------- |
| Latest         | `latest`   | The end of the topic. The task reads only new records.          |
| Earliest       | `earliest` | The oldest record that the topic keeps.                         |
| Committed only | `none`     | Nowhere. The task does not start until the group has an offset. |

Delivery is at least once, so a restart or a rebalance of the group can read a record again. After the task starts, and after the group assigns it partitions, it skips records with a timestamp at or before the last one in the index channel. On a topic with more than one partition, a replayed record that arrives after a newer one follows the rules in [timestamps](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#timestamps), so it can be written again.

## Task options

| Setting                            | Default | Notes                                                                                                               |
| ---------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------- |
| Topic `topic`                      |         | One topic for each task.                                                                                            |
| Consumer group `group`             | Empty   | Empty uses `kafka-<task key>`.                                                                                      |
| Start from `start_offset`          | Latest  | See [start offset](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#consumer-group-and-start-offset). |
| Data saving `data_saving_disabled` | On      | See [data saving](https://docs.synnaxlabs.com/reference/hardware/tasks#data-saving).                                |
| Auto start `auto_start`            | Off     | Starts the task each time the Core starts.                                                                          |

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

## Record and field reference

A record has these settings:

| Setting                 | Default         | Notes                                                  |
| ----------------------- | --------------- | ------------------------------------------------------ |
| Record key `record_key` | Any key         | The Kafka key of the records. One record for each key. |
| Source `index`          | Kafka timestamp | The key of the timestamp field. Empty is Kafka time.   |

A field has these settings:

| Setting                    | Default   | Notes                                           |
| -------------------------- | --------- | ----------------------------------------------- |
| Pointer `pointer`          |           | The JSON pointer to the value. Cannot be empty. |
| Data type `data_type`      | `float64` | Must match the data type of the channel.        |
| Time format `time_format`  | None      | Required for a `timestamp` channel.             |
| Channel `channel`          |           | The channel that records the value.             |
| Enum mapping `enum_values` | None      | Maps text labels in the record to numbers.      |

All fields of a record write to channels on one index channel, and one index channel belongs to one record.

## Troubleshooting

* [Skipped ... of ... records](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#skipped-of-records)

  * Cause

    The task read some records that it cannot record. The description gives the reason for the last one, with its partition and offset:

    * **Record is not JSON** -> the record value is not JSON.
    * **Record has no value at** -> the JSON value has nothing at the pointer of a field.
    * **Value at** -> the value does not fit the data type of the channel.
    * **Record stamped … is not after the last written** -> a record value timestamp is older than the last one in the index channel.

    The warning clears 5 seconds after the last skip.

  * Fix

    Check the pointers and data types against a record from the topic. Pointers start with `/` and are case sensitive.

* [Lost connection to cluster](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#lost-connection-to-cluster)

  * Cause

    The Core cannot reach the cluster while the task runs. The task keeps trying, and the status changes to “Reconnected to cluster” when records arrive again. The task then continues from its committed offset.

* [consumer group ... has no committed offset on topic ...](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#consumer-group-has-no-committed-offset-on-topic)

  * Cause

    **Start from** is **Committed only**, and the consumer group has no committed offset on the topic.

  * Fix

    Set **Start from** to **Latest** or **Earliest**, or set **Consumer group** to a group that has read the topic.

* [record key ...: claimed by more than one record](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#record-key-claimed-by-more-than-one-record)

  * Cause

    Two records of the task have the same record key.

  * Fix

    Give each record its own key, or move the fields into one record.

* [record key ...: channels must share one index](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#record-key-channels-must-share-one-index)[record keys ... and ...: channels on one index belong in one record](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#record-key-channels-must-share-one-index)

  * Cause

    The channels of a record use more than one index channel, or two records write to channels on the same index channel.

  * Fix

    Give each record channels on one index channel of its own.

* [field ...: data type ... does not match channel ...](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#field-data-type-does-not-match-channel)

  * Cause

    The data type of the field differs from the data type of the channel.

  * Fix

    Set the data type of the field to the data type of the channel.

* [Failed to fetch records](https://docs.synnaxlabs.com/reference/hardware/kafka/read-task#failed-to-fetch-records)

  * Cause

    The cluster refused to send records, for example because the user cannot read the topic. The task stops. The description gives the reason.

  * Fix

    Check the topic name and the ACLs of the user, then start the task again.

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