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.

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 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

In the devices toolbar, right-click the cluster and select Create read task. You can also run the “Create Kafka read task” command.
Enter the Topic, such as sensor-readings.
Click Add record, then set Record key, or leave it empty for any key.
Click Add field on the record, then set Pointer and Data type.
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.

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

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 does not create channels. Create one index channel for each record first.

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. 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, 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.
Data saving `data_saving_disabled`	On	See 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
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
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 ...
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
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 record keys ... and ...: channels on one index belong in one record
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 ...
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
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.

Connect to a cluster

Write task