Skip to main content
The Kafka ingress integration requires Restate v1.8.1 or later.
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, it resumes from the offsets the old subscription committed. Since Restate v1.8.1, deleting a subscription drains the records it already read and logs the last ingested offset of every topic partition. By checking these offsets before you start the integration, you hand over the topic with no record processed twice and none skipped.
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:
If the group.id option was not set when the subscription was created, the group id is the subscription id.
2

Delete the old subscription and wait for it to drain

Delete the existing subscription. Restate stops reading new records and waits up to 5 minutes for the records it already read to be ingested.
Wait until every Restate node logs Kafka subscription drained. Before that, each partition logs one of the following lines:
  • Kafka partition drained, last offset ingested into Restate: <offset>
  • Kafka partition drained, no records were ingested since the partition was assigned
  • Draining the Kafka partition failed, together with its last ingested offset
Restate also logs the committed consumer group offsets in lines starting with Committed Kafka consumer group offsets of the draining subscription. A partition can appear in several of these lines. The last one that mentions it shows its committed offset.
3

Check the committed offsets

For every partition that logged a last ingested offset, the committed offset must be the last ingested offset + 1, because a committed offset is the next offset to read.If the offsets do not match, the partition has no commit line, or the commit line reports an error for it, set the consumer group offset of that partition before starting the integration:
Partitions that logged no records were ingested can keep their current offset.
4

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.
Draining is best effort. If the server crashes, or the drain does not finish within 5 minutes (logged as did not finish draining), partitions without a drained or failed line may have had records in flight. Starting the integration from an older offset then re-delivers those records as duplicate invocations, and starting from a newer one skips them. To avoid this, you can retry the drain by recreating the subscription under the same id with restate subscriptions create --id <subscription id>. See the Restate v1.8.1 release notes for the full procedure.