Kafka

Invoke handlers via Kafka events.

Connect your Restate handlers to Kafka topics. Restate takes care of Kafka consumer management and pushes events to your Restate handlers.

You get zero-overhead consumer management with automatic retries, durable execution, and stateful processing capabilities.

Each event 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.

Invoking Handlers via Kafka Events

You can invoke handlers via Kafka events, by doing the following:

1. Develop and register an event handler

   You can invoke any handler via Kafka events. The event payload will be (de)serialized as JSON.

   • When invoking Virtual Object or Workflow handlers via Kafka, the key of the Kafka record will be used to determine the Virtual Object/Workflow key. The key needs to be a valid UTF-8 string. The events 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 via Kafka, the key of the Kafka record will be used to determine the Virtual Object/Workflow key. The key needs to be a valid UTF-8 string. The events are delivered to the subscribed handler in parallel without ordering guarantees.
   • When invoking Service handlers over Kafka, events are delivered in parallel without ordering guarantees.

   Since you can invoke any handler via Kafka events, a single handler can be invoked both by RPC and via Kafka.

2. Register the Kafka cluster in Restate

   Register the Kafka cluster that Restate needs to connect to, using the Restate CLI:

   restate kafka-clusters create my-cluster bootstrap.servers=broker:9092
   

   You can pass any librdkafka configuration parameter as additional key=value arguments. Alternatively, you can read the properties from a file with -f my-cluster.properties, or open an editor on a properties template with --edit.

   Using SASL/SSL (e.g. Confluent Kafka)

   To connect to a Kafka cluster that requires SASL/SSL authentication (e.g., Confluent Kafka), you can specify the necessary parameters when registering the cluster:

   restate kafka-clusters create my-kafka \
       bootstrap.servers=my-kafka:9092 \
       security.protocol=SASL_SSL \
       sasl.mechanisms=PLAIN \
       sasl.username=user \
       sasl.password=pass \
       client.id=client-id
   

   For Confluent Cloud, you can copy the client configuration properties from the Confluent Cloud console into a properties file and register the cluster with it:

   restate kafka-clusters create my-kafka -f confluent-cloud.properties
   

   Using SASL OAuth2.0 / OpenID Connect

   The Kafka ingress supports SASL OAUTHBEARER authentication, enabling OAuth 2.0/OpenID Connect (OIDC) token-based connections to managed Kafka services.

   Configure SASL OAUTHBEARER via the cluster properties. These options are passed directly to librdkafka.

   Example for Confluent Cloud:

   restate kafka-clusters create my-cluster \
       bootstrap.servers=pkc-xxxxxx.eu-central-1.aws.confluent.cloud:9092 \
       security.protocol=SASL_SSL \
       sasl.mechanisms=OAUTHBEARER \
       sasl.oauthbearer.method=oidc \
       sasl.oauthbearer.client.id=<your-client-id> \
       sasl.oauthbearer.client.secret=<your-client-secret> \
       sasl.oauthbearer.token.endpoint.url=<your-token-endpoint> \
       sasl.oauthbearer.scope=kafka
   

   Common OAUTHBEARER options:

   Option	Description
   security.protocol	Set to SASL_SSL for encrypted connections
   sasl.mechanism	Set to OAUTHBEARER
   sasl.oauthbearer.method	Set to oidc for OIDC-based token retrieval
   sasl.oauthbearer.client.id	OAuth client ID
   sasl.oauthbearer.client.secret	OAuth client secret
   sasl.oauthbearer.token.endpoint.url	OAuth token endpoint URL
   sasl.oauthbearer.scope	OAuth scope (if required by provider)

   For the full list of available options, see the librdkafka CONFIGURATION.md.

3. Register the service you want to invoke.
4. Subscribe the event handler to the Kafka topic

   Let Restate forward events from the Kafka topic to the event handler by creating a subscription:

   restate subscriptions create kafka://my-cluster/my-topic service://MyService/handle auto.offset.reset=earliest
   

   Once you’ve created a subscription, Restate immediately starts consuming events from Kafka. The handler will be invoked for each event received from Kafka.

   The trailing key=value options are optional and accept any configuration parameter from librdkafka configuration.

Kafka connection configuration

You can pass arbitrary Kafka cluster properties when registering the cluster, and those properties will be applied for all the subscriptions to that cluster, for example:

restate kafka-clusters create my-cluster \
    bootstrap.servers=broker:9092 \
    sasl.username=me \
    sasl.password=pass

For the full list of options, check librdkafka configuration.

Multiple Kafka clusters support

You can register multiple Kafka clusters:

restate kafka-clusters create my-cluster-1 bootstrap.servers=localhost:9092
restate kafka-clusters create my-cluster-2 bootstrap.servers=localhost:9093

And then, when creating the subscriptions, you refer to the specific cluster by name:

# Subscription to my-cluster-1
restate subscriptions create kafka://my-cluster-1/topic-1 service://MyService/handleCluster1

# Subscription to my-cluster-2
restate subscriptions create kafka://my-cluster-2/topic-2 service://MyService/handleCluster2

Event metadata

You can access the event metadata in the handler by getting the request headers map:

TypeScript

ctx.request().headers,

Each event carries within this map the following entries:

• restate.subscription.id: The subscription identifier, as shown by restate subscriptions list.
• kafka.offset: The record offset.
• kafka.partition: The record partition.
• kafka.timestamp: The record timestamp.

Raw event support

Check out the serialization documentation of your SDK to learn how to receive raw events in your handler.

Managing Kafka Clusters

Manage the registered Kafka clusters with the restate kafka-clusters CLI commands (alias kc):

# List the registered Kafka clusters
restate kafka-clusters list
# Print the properties of a Kafka cluster and its subscriptions
restate kafka-clusters describe my-cluster
# Open an editor to interactively update the cluster properties
restate kafka-clusters edit my-cluster
# Update the cluster properties non-interactively, for CI and scripting
restate kafka-clusters patch my-cluster --set client.id=restate-cli --unset sasl.password
# Remove a Kafka cluster
restate kafka-clusters delete my-cluster

Managing Kafka Subscriptions

Manage the subscriptions with the restate subscriptions CLI commands (alias sub):

# Subscribe a handler to a Kafka topic
restate subscriptions create kafka://my-cluster/my-topic service://MyService/Handle auto.offset.reset=earliest
# List current subscriptions
restate subscriptions list
# Print detailed information about a subscription, including its options
restate subscriptions describe sub_11XHoawrCiWtv8kzhEyGtsR
# Remove a subscription using its ID (starts with sub_)
restate subscriptions delete sub_11XHoawrCiWtv8kzhEyGtsR

When you delete a subscription, Restate stops the associated consumer group. Messages already enqueued by Restate will still be processed.