Skip to main content
The Kafka ingress integration connects Apache Kafka to Restate. It consumes records from your topics and turns each one into an invocation, so you can process Kafka events with durable execution, automatic retries, and stateful handlers, without writing any consumer glue code. The integration runs as its own container and scales independently of the Restate server.
Each record leads to an invocation, meaning a request to execute a handler. Each invocation has its own unique ID and lifecycle. Have a look at managing invocations to learn how to manage the lifecycle of an invocation.
This integration is the recommended way to consume Kafka events and replaces the old Kafka support. If you are already using the old support, see Migrating from the old Kafka support.

​
How it works

The integration runs as a container and does the following:
  • Consumes the configured topics using a Kafka consumer group.
  • Maps each record to an invocation using a record mapper.
  • Pushes the invocation to the configured Restate.
What you get is:
  • Exactly-once processing. Restate deduplicates based on consumer group id, topic, partition and offset, so restarts and rebalances never cause duplicate or lost invocations.
  • Per-partition ordering. Records from a single partition are always delivered in order.

​
Prerequisites

  • A running Restate server, and its ingress URL.
  • A Kafka cluster.
  • The service you want to invoke, deployed and registered with Restate.

​
Getting started

1

Develop and register an event handler

You can invoke any handler from a Kafka record. The record value is (de)serialized as JSON.
  • When invoking Virtual Object or Workflow handlers, the key of the Kafka record is used to determine the Virtual Object or Workflow key. The key needs to be a valid UTF-8 string. Records are delivered to the subscribed handler in the order in which they arrived on the topic partition.
  • When invoking Virtual Object or Workflow shared handlers, the key of the Kafka record is used to determine the Virtual Object or Workflow key. The key needs to be a valid UTF-8 string. Records are delivered in parallel without ordering guarantees.
  • When invoking Service handlers, records are delivered in parallel without ordering guarantees.
Since you can invoke any handler, a single handler can be invoked both by RPC and from Kafka records.
2

Run the integration

Start the integration container, pointing it at your Kafka cluster, your topics, and your Restate ingress:
The example above uses the static record mapper, which routes every record on the subscribed topics to the same service and handler.
3

Process events

Once the integration is running, it immediately starts consuming records from the configured topics. The handler is invoked for each record, and you can follow the resulting invocations in the Restate UI or with the CLI.

​
Configuration

Configure the integration with environment variables, or with a .properties file referenced by CONFIG_FILE. Environment variables take precedence. For the full list of options with their defaults, see the configuration reference.
Routes every record on the subscribed topics to the same service and handler. The Kafka record key becomes the Virtual Object or Workflow key, and the record value becomes the request payload.The example below sends records from the orders and payments topics to the OrderService/onKafkaEvent handler:
Derives the target service and handler from each record value using JSON Pointers. Use it when a single topic carries different event types that need different handlers.The example below consumes the orders topic, picks a handler on the Order service from each record’s type field, keys the invocation by customerId, and deduplicates on eventId:
Any additional KAFKA_* environment variable is forwarded to the Kafka consumer, mapping underscores to the Apache Kafka client.The example below sends records from the orders topic to the OrderService/onKafkaEvent handler over a SASL/SSL connection:

​
Record metadata

When the record mapper attaches Kafka metadata, you can read it in the handler through the request headers map:
Each record carries within this map the following entries:
  • kafka.topic: The topic the record was consumed from.
  • kafka.partition: The record partition.
  • kafka.offset: The record offset.
  • kafka.timestamp: The record timestamp.

​
Migrating from the old Kafka support

The old Kafka support runs the consumer inside the Restate server and is configured with the restate kafka-clusters and restate subscriptions CLI commands. This integration runs as a separate container instead, so migrating means moving that configuration into the container and stopping the old subscription. Kafka tracks committed offsets per consumer group, so if the integration uses the same group.id as your existing subscription, the new consumer resumes exactly where the old one stopped, with no gap and no reprocessing.
1

Find the group.id of your existing subscription

Identify the group.id your current subscription uses. List your subscriptions, then describe the one you want to migrate to see its options, including the group.id:
You can set the group.id explicitly with the group.id option when creating a subscription. Otherwise Restate assigns one.
2

Stop the old subscription

Delete the existing subscription so the old consumer stops. Its committed offsets remain in Kafka under the group id.
3

Start the integration with the same group.id

Run the integration with KAFKA_GROUP_ID set to that same group id, and the same topic and target handler:
Do not run the old subscription and the integration against the same topic with the same group.id at the same time. They would join the same consumer group and split partitions between them, mixing the two delivery semantics. Stop the old subscription before starting the integration.