Event Processing

Kafka event processing requires managing Kafka consumers, handling retries, maintaining state stores, and coordinating complex workflows. Restate eliminates this complexity by providing lightweight, transactional event processing with zero consumer management and built-in state.

Workflows from Kafka

Build event handlers with complex control flow, loops, timers, and transactional guarantees:

export default restate.object({
  name: "userFeed",
  handlers: {
    processPost: async (ctx: restate.ObjectContext, post: SocialMediaPost) => {
      const userId = ctx.key;

      // Durable side effect: persisted and replayed on retries
      const postId = await ctx.run(() => createPost(userId, post));

      // Wait for processing to complete with durable timers
      while ((await ctx.run(() => getPostStatus(postId))) === PENDING) {
        await ctx.sleep({ seconds: 5 });
      }

      await ctx.run(() => updateUserFeed(userId, postId));
    },
  },
});

@VirtualObject
public class UserFeed {
  @Handler
  public void processPost(ObjectContext ctx, SocialMediaPost post) {
    String userId = ctx.key();

    String postId = ctx.run(String.class, () -> createPost(userId, post));

    while (ctx.run(String.class, () -> getPostStatus(postId)).equals("PENDING")) {
      ctx.sleep(Duration.ofSeconds(5));
    }

    ctx.run(() -> updateUserFeed(userId, postId));
  }
}

user_feed = restate.VirtualObject("UserFeed")


@user_feed.handler()
async def process_post(ctx: restate.ObjectContext, post: SocialMediaPost):
    user_id = ctx.key()

    post_id = await ctx.run_typed(
        "create post", create_post, user_id=user_id, post=post
    )

    while (
        await ctx.run_typed("get status", get_post_status, post_id=post_id)
        == Status.PENDING
    ):
        await ctx.sleep(timedelta(seconds=5))

    await ctx.run_typed("update feed", update_user_feed, user=user_id, post_id=post_id)

type UserFeed struct{}

func (UserFeed) ProcessPost(ctx restate.ObjectContext, post SocialMediaPost) error {
  var userId = restate.Key(ctx)

  postId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
    return CreatePost(userId, post)
  })
  if err != nil {
    return err
  }

  for {
    status, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
      return GetPostStatus(postId), nil
    })
    if err != nil {
      return err
    }
    if status != PENDING {
      break
    }
    if err = restate.Sleep(ctx, 5*time.Second); err != nil {
      // </mark_2>
      return err
    }
  }

  if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
    return UpdateUserFeed(userId, postId)
  }); err != nil {
    return err
  }

  return nil
}

Key Benefits:

Push-based delivery: Events delivered directly to handlers with zero consumer management
Durable execution: Failed handlers are retried with exponential backoff until they succeed. Handlers replay previously completed steps and resume exactly where they left off.
Parallel processing: Events for different keys process concurrently, like a queue per object key
Timers and scheduling: Timers and delays that survive crashes and restarts for long-running workflows

Stateful Event Handlers

Maintain K/V state across events:

export default restate.object({
  name: "delivery-tracker",
  handlers: {
    register: async (ctx: restate.ObjectContext, delivery: Delivery) =>
      ctx.set("delivery", delivery),

    setLocation: async (ctx: restate.ObjectContext, location: Location) => {
      const delivery = await ctx.get<Delivery>("delivery");
      if (!delivery) {
        throw new TerminalError(`Delivery not found`);
      }

      delivery.locations.push(location);
      ctx.set("delivery", delivery);
    },

    getDelivery: shared(async (ctx: restate.ObjectSharedContext) =>
      ctx.get<Delivery>("delivery")
    ),
  },
});

@VirtualObject
public class DeliveryTracker {
  private static final StateKey<Delivery> DELIVERY = StateKey.of("delivery", Delivery.class);

  @Handler
  public void register(ObjectContext ctx, Delivery packageInfo) {
    ctx.set(DELIVERY, packageInfo);
  }

  @Handler
  public void setLocation(ObjectContext ctx, Location location) {
    var delivery = ctx.get(DELIVERY).orElseThrow(() -> new TerminalException("Delivery not found"));

    delivery.addLocation(location);
    ctx.set(DELIVERY, delivery);
  }

  @Shared
  public Delivery getDelivery(SharedObjectContext ctx) {
    return ctx.get(DELIVERY).orElseThrow(() -> new TerminalException("Delivery not found"));
  }
}

delivery_tracker = restate.VirtualObject("DeliveryTracker")


@delivery_tracker.handler()
async def register(ctx: restate.ObjectContext, delivery: Delivery):
    ctx.set("delivery", delivery)


@delivery_tracker.handler()
async def set_location(ctx: restate.ObjectContext, location: Location):
    delivery = await ctx.get("delivery", type_hint=Delivery)
    if delivery is None:
        raise TerminalError(f"Delivery {ctx.key()} not found")

    delivery.locations.append(location)
    ctx.set("delivery", delivery)


@delivery_tracker.handler(kind="shared")
async def get_delivery(ctx: restate.ObjectSharedContext) -> Delivery:
    delivery = await ctx.get("delivery", type_hint=Delivery)
    if delivery is None:
        raise TerminalError(f"Delivery {ctx.key()} not found")
    return delivery

type DeliveryTracker struct{}

func (DeliveryTracker) Register(ctx restate.ObjectContext, delivery Delivery) error {
  restate.Set[Delivery](ctx, "delivery", delivery)
  return nil
}

func (DeliveryTracker) SetLocation(ctx restate.ObjectContext, location Location) error {
  packageInfo, err := restate.Get[*Delivery](ctx, "delivery")
  if err != nil {
    return err
  }
  if packageInfo == nil {
    return restate.TerminalError(errors.New("delivery not found"))
  }

  packageInfo.Locations = append(packageInfo.Locations, location)
  restate.Set[Delivery](ctx, "delivery", *packageInfo)
  return nil
}

func (DeliveryTracker) GetDelivery(ctx restate.ObjectSharedContext) (*Delivery, error) {
  return restate.Get[*Delivery](ctx, "delivery")
}

Key Benefits:

Persistent state: Store and retrieve state directly in handlers without external stores
Built-in consistency: State operations are always consistent with execution
Agents, actors, digital twins: Model stateful entities that react to events

When to Choose Restate

✅ Choose Restate when you need:

Kafka integration: Process Kafka events with zero consumer management
Reliable processing: Automatic retry and recovery for failed event handlers
Transactional processing: Execute side effects with durable execution guarantees
Stateful event processing: Maintain state across events without external stores
Event-driven workflows: Build complex flows with loops, timers, and conditions

Processing events with Restate? Join our community on Discord or Slack to discuss your use case.

Comparison with Other Solutions

Feature	Restate	Traditional Kafka Processing	Stream Processing Frameworks
Event Delivery	Push-based to durable handlers	Polling-based consumer groups	Built-in sources and sinks
Consumer Management	Zero configuration	Manual offset and consumer group management	Zero configuration
Failure Recovery	Fine-grained progress persistence with durable side effects	Coarse-grained offset commits with potential reprocessing	Coarse-grained checkpointing with potential duplicate side effects
State Management	Built-in durable state	External state store required	Built-in state store
Queuing Semantics	Queue-per-key with ordering guarantees	Queue-per-partition with ordering guarantees	Queue-per-partition (inherited from Kafka)
Complex Workflows	Unlimited control flow: loops, timers, conditions	Long-running logic blocks consumer loop; requires external services	DAG-based processing with limited control flow
Best For	Event-driven state machines, transactional processing, complex workflows	Simple ETL and message processing	High-throughput analytics, aggregations, joins

Getting Started

Ready to build event processing systems with Restate? Here are your next steps:

Quickstart

Set up Restate and process your first events

Kafka Quickstart

Follow the quickstart to implement your first durable event handler

Examples

Explore templates, patterns, and end-to-end applications

Evaluating Restate and missing a feature? Contact us on Discord or Slack.

Use Cases

Foundations

Tour of Restate for...

Event Processing

Workflows from Kafka

Stateful Event Handlers

When to Choose Restate

Comparison with Other Solutions

Getting Started

Quickstart

Kafka Quickstart

Examples

​Workflows from Kafka

​Stateful Event Handlers

​When to Choose Restate

​Comparison with Other Solutions

​Getting Started

Quickstart

Kafka Quickstart

Examples

Workflows from Kafka

Stateful Event Handlers

When to Choose Restate

Comparison with Other Solutions

Getting Started