Kafka Message Client

Configure AnyLog to consume Kafka topics and map Kafka messages into AnyLog tables.

Overview

AnyLog can subscribe to Kafka topics and ingest Kafka messages into local tables using the same mapping model used by MQTT message clients. The Kafka consumer reads messages from a Kafka broker, applies the configured topic mapping, and sends the resulting rows through AnyLog’s normal streaming and storage workflow.

Use Kafka when the source system already publishes to Kafka, when Kafka is the site’s event backbone, or when Kafka-specific retention and consumer-group behavior are part of the deployment. Use MQTT when publishers are lightweight devices or MQTT brokers are already deployed at the edge.

Run a Kafka Consumer

<run kafka consumer where
    ip = [ip] and
    port = [port] and
    reset = [latest|earliest] and
    topic = (
        name = [topic] and
        dbms = [dbms mapping] and
        table = [table mapping] and
        column.[name].[type] = [value mapping]
    )>

Options

Option	Description	Default
`ip`	Kafka broker IP address.
`port`	Kafka broker port.
`reset`	Offset policy. Use `latest` to consume new messages only, or `earliest` to consume available retained messages.	`latest`
`topic`	One or more topic blocks with mapping instructions.

Topic mapping supports the same dbms, table, column.[name].[type], and bring expressions used by MQTT message clients. See MQTT Message Broker for the JSON mapping syntax.

Example

<run kafka consumer where
    ip = [ip] and
    port = [port] and
    reset = latest and
    topic = (
        name = my-data and
        dbms = "bring [dbms]" and
        table = "bring [sensor]" and
        column.timestamp.timestamp = "bring [timestamp]" and
        column.value.float = "bring [value]"
    )>

After data is ingested, query it through the AnyLog network:

run client () sql [dbms] format=table and extend=(+ip, +node_name, @table_name) "select * from [table] limit 10"

Local Kafka for development

Use a local Docker-based Kafka broker for testing consumer mappings without a production cluster.

Start the broker

docker run -d --rm --name kafka-dev -p 9092:9092 apache/kafka:latest

Use localhost:9092 as --bootstrap-server when running Kafka CLI commands inside the container via docker exec. Use the broker machine’s LAN IP (e.g. 192.168.1.101:9092) when connecting from another host.

Create a topic

docker exec kafka-dev /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create --if-not-exists \
  --topic test --partitions 1 --replication-factor 1

Publish a message

echo '{"timestamp":1776294106000,"value":42.0,"deviceID":"d1"}' | \
docker exec -i kafka-dev /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic test

Verify messages

docker exec kafka-dev /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test \
  --group "tmp-$(date +%s)" \
  --consumer-property auto.offset.reset=earliest \
  --max-messages 5

Connect AnyLog and start consuming

On the Operator node, connect the database and start the consumer:

connect dbms new_company where type = sqlite

<run kafka consumer where ip = localhost and
    port = 9092 and
    reset = earliest and
    topic = (name = test and
        dbms = new_company and
        table = kafka_demo and
        column.timestamp.timestamp = "bring [timestamp]" and
        column.value.float = "bring [value]" and
        column.deviceid.str = "bring [deviceID]")
>

Topic mapping for Kafka uses the same JSON/bring model as MQTT — see MQTT Message Broker and Mapping Policy.

Verify data is flowing:

get streaming
sql new_company "select * from kafka_demo"

Stop the broker

docker stop kafka-dev

The --rm flag on docker run removes the container automatically when it stops.

Notes

Unlike MQTT’s broker = local shorthand, Kafka has no AnyLog-local broker. For local testing, run a real Kafka endpoint (see Local Kafka for development) and point ip / port at it (for example localhost and 9092).

For deployments that use both MQTT and Kafka, keep the topic-to-table naming rules consistent across consumers so queries can target predictable AnyLog tables.