Event Processing

Lightweight, transactional event processing.

Process Kafka events with flexible flows of transactional steps.
Restate takes care of the event plumbing and pushes events to your handler.

Lightweight Durable Functions

Functions are executed durably with fine-grained retries and workflow-like semantics.

Queue per key

Restate splits partitions into a queue per key. If it takes long to process an event, the events for other keys aren’t blocked.

Push events to functions

Restate pushes events to your handlers. No need to implement consumers, manage subscriptions/offsets, etc.

Event processing with Restate

Connect functions to Kafka topics. Restate pushes the events to your function.

TypeScript
Java
Kotlin
Go

Lightweight Durable Functions

Write functions that take Kafka events as inputs. Functions execute with Durable Execution: their progress is tracked and they can be retried from the exact point before the crash, as if you are taking micro-checkpoints throughout the function execution.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

subscribe.sh


curl restate:9070/subscriptions --json '{
    "source": "kafka://my-cluster/user-events",
    "sink": "service://userUpdates/updateUserEvent"
}'

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Lightweight Durable Functions

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

user_updates.ts


const userUpdates = restate.object({
  name: "userUpdates",
  handlers: {
    updateUserEvent: async (ctx: restate.ObjectContext, event: UserUpdate) => {
      const { profile, permissions, resources } = verifyEvent(event);
      let userId = await ctx.run(() => updateProfile(profile));
      while (userId === NOT_READY) {
        await ctx.sleep(5_000);
        userId = await ctx.run(() => updateProfile(profile));
      }
      const roleId = await ctx.run(() => setPermissions(userId, permissions));
      await ctx.run(() => provisionResources(userId, roleId, resources));
    },
  },
});

Lightweight Durable Functions

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

subscribe.sh


curl restate:9070/subscriptions --json '{
"source": "kafka://my-cluster/user-events",
"sink": "service://UserUpdatesService/updateUserEvent"
}'

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Lightweight Durable Functions

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

UserUpdatesService.java


@VirtualObject
public class UserUpdatesService {
  @Handler
  public void updateUserEvent(ObjectContext ctx, UserUpdate event) {
    String userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    while (userId.equals("NOT_READY")) {
      ctx.sleep(Duration.ofMillis(5000));
      userId = ctx.run(JsonSerdes.STRING, () -> updateUserProfile(event.getProfile()));
    }
    String finalUserId = userId;
    String roleId =
        ctx.run(JsonSerdes.STRING, () -> setUserPermissions(finalUserId, event.getPermissions()));
    ctx.run(() -> provisionResources(finalUserId, roleId, event.getResources()));
  }
}

Lightweight Durable Functions

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

subscribe.sh


curl restate:9070/subscriptions --json '{
"source": "kafka://my-cluster/user-events",
"sink": "service://UserUpdatesService/updateUserEvent"
}'

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Lightweight Durable Functions

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

UserUpdatesService.kt


@VirtualObject
class UserUpdatesService {
  @Handler
  suspend fun updateUserEvent(ctx: ObjectContext, event: UserUpdate) {
    var userId = ctx.runBlock { updateUserProfile(event.profile) }
    while (userId == "NOT_READY") {
      ctx.sleep(5000.milliseconds)
      userId = ctx.runBlock { updateUserProfile(event.profile) }
    }
    val finalUserId = userId
    val roleId = ctx.runBlock { setUserPermissions(finalUserId, event.permissions) }
    ctx.runBlock { provisionResources(finalUserId, roleId, event.resources) }
  }
}

Lightweight Durable Functions

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

subscribe.sh


curl restate:9070/subscriptions --json '{
    "source": "kafka://my-cluster/user-events",
    "sink": "service://userUpdates/updateUserEvent"
}'

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

Lightweight Durable Functions

Push events to functions

Let Restate subscribe to a Kafka topic and specify to which function to push the events. Restate will take care of the event plumbing: polling for records, committing offsets, recovering...

Queue per key

Events get sent to objects based on the Kafka key. For each key, Restate ensures that events are processed sequentially and in order. Slow events on other keys do not block processing.

In the example, we process user updates in a queue per user. Slow updates for one user do not block updates for other users.

Postpone processing

Flexibly postpone processing of events until later. Restate tracks the timers and re-invokes. When sleeping, other events for that key are enqueued.

Here, we postpone processing for 5 seconds if the user profile is not ready yet.

Durable side effects

The results of interactions with external systems are tracked and recovered after failures. This simplifies writing flows that keep multiple systems in sync.

Flexible control flow

As opposed to many stream processing systems, Restate does not put any restrictions on the control flow (e.g. DAG). Each event crafts its own path through the code and builds up its own recovery log.

userupdates.go


func (UserUpdates) UpdateUserEvent(ctx restate.ObjectContext, event UserUpdate) error {
	userId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return updateProfile(ctx, event.Profile)
	})
	if err != nil {
		return err
	}
	for userId == NOT_READY {
		if err := restate.Sleep(ctx, 5*time.Second); err != nil {
			return err
		}
		userId, err = restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
			return updateProfile(ctx, event.Profile)
		})
		if err != nil {
			return err
		}
	}
	roleId, err := restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		return setPermissions(ctx, userId, event.Permissions)
	})
	if err != nil {
		return err
	}
	if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
		return restate.Void{}, provisionResources(ctx, userId, roleId, event.Resources)
	}); err != nil {
		return err
	}
	return nil
}

LOW-LATENCY

Restate’s event-driven foundation built in Rust lets you process events at high speed. Restate keeps a queue per key and pushes events to your functions to process them in parallel as fast as possible.

DURABLE EXECUTION

Restate manages the complexities of reading from Kafka to make sure each event gets processed exactly once. Restate handles retries and recovery of your event handlers to the exact point before the crash.

Stateful event processing with Restate

Implement stateful event handlers with Restate.

TypeScript
Java
Kotlin
Go

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

profile_service.ts


const eventEnricher = restate.object({
  name: "profile",
  handlers: {
    userEvent: async (ctx: ObjectContext, event: UserProfile) => {
      ctx.set("user", event);
      ctx.objectSendClient(EventEnricher, ctx.key).emit(restate.rpc.sendOpts({ delay: 1000 }));
    },
    featureEvent: async (ctx: ObjectContext, featureEvent: string) => {
      const userEvent = await ctx.get<UserProfile>("user");
      (userEvent!.features ??= []).push(featureEvent);
      ctx.set("user", userEvent);
    },
    emit: async (ctx: ObjectContext) => {
      const user = await ctx.get<UserProfile>("user")
      send(ctx.key, user);
      ctx.clearAll();
    },
  },
});
type EventEnricherType = typeof eventEnricher;
const EventEnricher: EventEnricherType = { name: "profile" };

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

profile_service.ts


const eventEnricher = restate.object({
  name: "profile",
  handlers: {
    userEvent: async (ctx: ObjectContext, event: UserProfile) => {
      ctx.set("user", event);
      ctx.objectSendClient(EventEnricher, ctx.key).emit(restate.rpc.sendOpts({ delay: 1000 }));
    },
    featureEvent: async (ctx: ObjectContext, featureEvent: string) => {
      const userEvent = await ctx.get<UserProfile>("user");
      (userEvent!.features ??= []).push(featureEvent);
      ctx.set("user", userEvent);
    },
    emit: async (ctx: ObjectContext) => {
      const user = await ctx.get<UserProfile>("user")
      send(ctx.key, user);
      ctx.clearAll();
    },
  },
});
type EventEnricherType = typeof eventEnricher;
const EventEnricher: EventEnricherType = { name: "profile" };

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

profile_service.ts


const eventEnricher = restate.object({
  name: "profile",
  handlers: {
    userEvent: async (ctx: ObjectContext, event: UserProfile) => {
      ctx.set("user", event);
      ctx.objectSendClient(EventEnricher, ctx.key).emit(restate.rpc.sendOpts({ delay: 1000 }));
    },
    featureEvent: async (ctx: ObjectContext, featureEvent: string) => {
      const userEvent = await ctx.get<UserProfile>("user");
      (userEvent!.features ??= []).push(featureEvent);
      ctx.set("user", userEvent);
    },
    emit: async (ctx: ObjectContext) => {
      const user = await ctx.get<UserProfile>("user")
      send(ctx.key, user);
      ctx.clearAll();
    },
  },
});
type EventEnricherType = typeof eventEnricher;
const EventEnricher: EventEnricherType = { name: "profile" };

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

profile_service.ts


const eventEnricher = restate.object({
  name: "profile",
  handlers: {
    userEvent: async (ctx: ObjectContext, event: UserProfile) => {
      ctx.set("user", event);
      ctx.objectSendClient(EventEnricher, ctx.key).emit(restate.rpc.sendOpts({ delay: 1000 }));
    },
    featureEvent: async (ctx: ObjectContext, featureEvent: string) => {
      const userEvent = await ctx.get<UserProfile>("user");
      (userEvent!.features ??= []).push(featureEvent);
      ctx.set("user", userEvent);
    },
    emit: async (ctx: ObjectContext) => {
      const user = await ctx.get<UserProfile>("user")
      send(ctx.key, user);
      ctx.clearAll();
    },
  },
});
type EventEnricherType = typeof eventEnricher;
const EventEnricher: EventEnricherType = { name: "profile" };

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

profile_service.ts


const eventEnricher = restate.object({
  name: "profile",
  handlers: {
    userEvent: async (ctx: ObjectContext, event: UserProfile) => {
      ctx.set("user", event);
      ctx.objectSendClient(EventEnricher, ctx.key).emit(restate.rpc.sendOpts({ delay: 1000 }));
    },
    featureEvent: async (ctx: ObjectContext, featureEvent: string) => {
      const userEvent = await ctx.get<UserProfile>("user");
      (userEvent!.features ??= []).push(featureEvent);
      ctx.set("user", userEvent);
    },
    emit: async (ctx: ObjectContext) => {
      const user = await ctx.get<UserProfile>("user")
      send(ctx.key, user);
      ctx.clearAll();
    },
  },
});
type EventEnricherType = typeof eventEnricher;
const EventEnricher: EventEnricherType = { name: "profile" };

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

ProfileService.java


@VirtualObject
public class ProfileService {
  private static final StateKey<UserProfile> USER =
      StateKey.of("user", JacksonSerdes.of(UserProfile.class));
  @Handler
  public void userEvent(ObjectContext ctx, String name) {
    UserProfile profile = new UserProfile(ctx.key(), name);
    ctx.set(USER, profile);
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(Duration.ofSeconds(1)).emit();
  }
  @Handler
  public void featureEvent(ObjectContext ctx, String email) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    user.setEmail(email);
    ctx.set(USER, user);
  }
  @Handler
  public void emit(ObjectContext ctx) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    send(ctx.key(), user);
    ctx.clearAll();
  }
}

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

ProfileService.java


@VirtualObject
public class ProfileService {
  private static final StateKey<UserProfile> USER =
      StateKey.of("user", JacksonSerdes.of(UserProfile.class));
  @Handler
  public void userEvent(ObjectContext ctx, String name) {
    UserProfile profile = new UserProfile(ctx.key(), name);
    ctx.set(USER, profile);
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(Duration.ofSeconds(1)).emit();
  }
  @Handler
  public void featureEvent(ObjectContext ctx, String email) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    user.setEmail(email);
    ctx.set(USER, user);
  }
  @Handler
  public void emit(ObjectContext ctx) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    send(ctx.key(), user);
    ctx.clearAll();
  }
}

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

ProfileService.java


@VirtualObject
public class ProfileService {
  private static final StateKey<UserProfile> USER =
      StateKey.of("user", JacksonSerdes.of(UserProfile.class));
  @Handler
  public void userEvent(ObjectContext ctx, String name) {
    UserProfile profile = new UserProfile(ctx.key(), name);
    ctx.set(USER, profile);
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(Duration.ofSeconds(1)).emit();
  }
  @Handler
  public void featureEvent(ObjectContext ctx, String email) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    user.setEmail(email);
    ctx.set(USER, user);
  }
  @Handler
  public void emit(ObjectContext ctx) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    send(ctx.key(), user);
    ctx.clearAll();
  }
}

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

ProfileService.java


@VirtualObject
public class ProfileService {
  private static final StateKey<UserProfile> USER =
      StateKey.of("user", JacksonSerdes.of(UserProfile.class));
  @Handler
  public void userEvent(ObjectContext ctx, String name) {
    UserProfile profile = new UserProfile(ctx.key(), name);
    ctx.set(USER, profile);
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(Duration.ofSeconds(1)).emit();
  }
  @Handler
  public void featureEvent(ObjectContext ctx, String email) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    user.setEmail(email);
    ctx.set(USER, user);
  }
  @Handler
  public void emit(ObjectContext ctx) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    send(ctx.key(), user);
    ctx.clearAll();
  }
}

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

ProfileService.java


@VirtualObject
public class ProfileService {
  private static final StateKey<UserProfile> USER =
      StateKey.of("user", JacksonSerdes.of(UserProfile.class));
  @Handler
  public void userEvent(ObjectContext ctx, String name) {
    UserProfile profile = new UserProfile(ctx.key(), name);
    ctx.set(USER, profile);
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(Duration.ofSeconds(1)).emit();
  }
  @Handler
  public void featureEvent(ObjectContext ctx, String email) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    user.setEmail(email);
    ctx.set(USER, user);
  }
  @Handler
  public void emit(ObjectContext ctx) {
    UserProfile user =
        ctx.get(USER)
            .orElseThrow(() -> new TerminalException("No user found"));
    send(ctx.key(), user);
    ctx.clearAll();
  }
}

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

ProfileService.kt


@VirtualObject
class ProfileService {
  companion object {
    private val USER = StateKey.of("user", KtSerdes.json<UserProfile>())
  }
  @Handler
  suspend fun userEvent(ctx: ObjectContext, name: String) {
    val profile = UserProfile(ctx.key(), name)
    ctx.set(USER, profile)
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(1.seconds).emit()
  }
  @Handler
  suspend fun featureEvent(ctx: ObjectContext, email: String) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    user.email = email
    ctx.set(USER, user)
  }
  @Handler
  suspend fun emit(ctx: ObjectContext) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    send(ctx.key(), user)
    ctx.clearAll()
  }
}

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

ProfileService.kt


@VirtualObject
class ProfileService {
  companion object {
    private val USER = StateKey.of("user", KtSerdes.json<UserProfile>())
  }
  @Handler
  suspend fun userEvent(ctx: ObjectContext, name: String) {
    val profile = UserProfile(ctx.key(), name)
    ctx.set(USER, profile)
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(1.seconds).emit()
  }
  @Handler
  suspend fun featureEvent(ctx: ObjectContext, email: String) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    user.email = email
    ctx.set(USER, user)
  }
  @Handler
  suspend fun emit(ctx: ObjectContext) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    send(ctx.key(), user)
    ctx.clearAll()
  }
}

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

ProfileService.kt


@VirtualObject
class ProfileService {
  companion object {
    private val USER = StateKey.of("user", KtSerdes.json<UserProfile>())
  }
  @Handler
  suspend fun userEvent(ctx: ObjectContext, name: String) {
    val profile = UserProfile(ctx.key(), name)
    ctx.set(USER, profile)
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(1.seconds).emit()
  }
  @Handler
  suspend fun featureEvent(ctx: ObjectContext, email: String) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    user.email = email
    ctx.set(USER, user)
  }
  @Handler
  suspend fun emit(ctx: ObjectContext) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    send(ctx.key(), user)
    ctx.clearAll()
  }
}

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

ProfileService.kt


@VirtualObject
class ProfileService {
  companion object {
    private val USER = StateKey.of("user", KtSerdes.json<UserProfile>())
  }
  @Handler
  suspend fun userEvent(ctx: ObjectContext, name: String) {
    val profile = UserProfile(ctx.key(), name)
    ctx.set(USER, profile)
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(1.seconds).emit()
  }
  @Handler
  suspend fun featureEvent(ctx: ObjectContext, email: String) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    user.email = email
    ctx.set(USER, user)
  }
  @Handler
  suspend fun emit(ctx: ObjectContext) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    send(ctx.key(), user)
    ctx.clearAll()
  }
}

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

ProfileService.kt


@VirtualObject
class ProfileService {
  companion object {
    private val USER = StateKey.of("user", KtSerdes.json<UserProfile>())
  }
  @Handler
  suspend fun userEvent(ctx: ObjectContext, name: String) {
    val profile = UserProfile(ctx.key(), name)
    ctx.set(USER, profile)
    ProfileServiceClient.fromContext(ctx, ctx.key()).send(1.seconds).emit()
  }
  @Handler
  suspend fun featureEvent(ctx: ObjectContext, email: String) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    user.email = email
    ctx.set(USER, user)
  }
  @Handler
  suspend fun emit(ctx: ObjectContext) {
    val user = ctx.get(USER) ?: throw TerminalException("No user found")
    send(ctx.key(), user)
    ctx.clearAll()
  }
}

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

profileservice.go


func (Profile) UserEvent(ctx restate.ObjectContext, event UserProfile) {
	restate.Set(ctx, "user", event)
	restate.ObjectSend(ctx, "EventEnricher", restate.Key(ctx), "Emit").
		Send(restate.Void{}, restate.WithDelay(1*time.Second))
}
func (Profile) FeatureEvent(ctx restate.ObjectContext, featureEvent string) error {
	userEvent, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	userEvent.Features = append(userEvent.Features, featureEvent)
	restate.Set(ctx, "user", userEvent)
	return nil
}
func (Profile) Emit(ctx restate.ObjectContext) error {
	user, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	send(restate.Key(ctx), user)
	restate.ClearAll(ctx)
	return nil
}

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

profileservice.go


func (Profile) UserEvent(ctx restate.ObjectContext, event UserProfile) {
	restate.Set(ctx, "user", event)
	restate.ObjectSend(ctx, "EventEnricher", restate.Key(ctx), "Emit").
		Send(restate.Void{}, restate.WithDelay(1*time.Second))
}
func (Profile) FeatureEvent(ctx restate.ObjectContext, featureEvent string) error {
	userEvent, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	userEvent.Features = append(userEvent.Features, featureEvent)
	restate.Set(ctx, "user", userEvent)
	return nil
}
func (Profile) Emit(ctx restate.ObjectContext) error {
	user, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	send(restate.Key(ctx), user)
	restate.ClearAll(ctx)
	return nil
}

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

profileservice.go


func (Profile) UserEvent(ctx restate.ObjectContext, event UserProfile) {
	restate.Set(ctx, "user", event)
	restate.ObjectSend(ctx, "EventEnricher", restate.Key(ctx), "Emit").
		Send(restate.Void{}, restate.WithDelay(1*time.Second))
}
func (Profile) FeatureEvent(ctx restate.ObjectContext, featureEvent string) error {
	userEvent, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	userEvent.Features = append(userEvent.Features, featureEvent)
	restate.Set(ctx, "user", userEvent)
	return nil
}
func (Profile) Emit(ctx restate.ObjectContext) error {
	user, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	send(restate.Key(ctx), user)
	restate.ClearAll(ctx)
	return nil
}

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

profileservice.go


func (Profile) UserEvent(ctx restate.ObjectContext, event UserProfile) {
	restate.Set(ctx, "user", event)
	restate.ObjectSend(ctx, "EventEnricher", restate.Key(ctx), "Emit").
		Send(restate.Void{}, restate.WithDelay(1*time.Second))
}
func (Profile) FeatureEvent(ctx restate.ObjectContext, featureEvent string) error {
	userEvent, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	userEvent.Features = append(userEvent.Features, featureEvent)
	restate.Set(ctx, "user", userEvent)
	return nil
}
func (Profile) Emit(ctx restate.ObjectContext) error {
	user, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	send(restate.Key(ctx), user)
	restate.ClearAll(ctx)
	return nil
}

K/V State

Store state in Restate and access it from other handlers. Restate guarantees that it is consistent and persistent. The state gets delivered together with the request, so you operate on local state.

Event Enrichment

Enrich events with data from multiple sources by storing it in state and eventually exposing it via other functions.

Delayed actions

Schedule async follow-up tasks for now or for later. Restate tracks the timers and triggers them when the time comes.

Here, we wait one second for other user features to arrive before sending the event to downstream processing.

Combine Kafka and RPC

Functions can be called over RPC or Kafka without changing the code. In the example, the registration can come over Kafka, while the email gets called via HTTP.

profileservice.go


func (Profile) UserEvent(ctx restate.ObjectContext, event UserProfile) {
	restate.Set(ctx, "user", event)
	restate.ObjectSend(ctx, "EventEnricher", restate.Key(ctx), "Emit").
		Send(restate.Void{}, restate.WithDelay(1*time.Second))
}
func (Profile) FeatureEvent(ctx restate.ObjectContext, featureEvent string) error {
	userEvent, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	userEvent.Features = append(userEvent.Features, featureEvent)
	restate.Set(ctx, "user", userEvent)
	return nil
}
func (Profile) Emit(ctx restate.ObjectContext) error {
	user, err := restate.Get[UserProfile](ctx, "user")
	if err != nil {
		return err
	}
	send(restate.Key(ctx), user)
	restate.ClearAll(ctx)
	return nil
}

Event handlers as regular, lightweight functions

Let Restate subscribe to a Kafka topic and push events to your functions.
Your functions run as normal services in your existing infra.

What you can build with Event Processing and Restate

Kafka-triggered workflows

Handle the payment, request the order preparation, wait for driver acceptance callback, etc.

Digital twin pattern

A delivery service that responds to location updates arriving over Kafka. The order gets updated accordingly.

Event-based task scheduling

For each event that comes in, schedule some tasks for in the future.

Wondering about a specific use case?

Let’s brainstorm together on Discord

Developer Resources

Blog post

Event-driven apps where event-driven is an implementation detail.

Docs

Read the docs to learn more.

Need help?

Join the Restate Discord channel

Event Processing

Lightweight, transactional event processing.

Lightweight Durable Functions

Queue per key

Push events to functions

Event processing with Restate​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

Lightweight Durable Functions​

Push events to functions​

Queue per key​

Postpone processing​

Durable side effects​

Flexible control flow​

LOW-LATENCY

DURABLE EXECUTION

Stateful event processing with Restate​

K/V State​

Event Enrichment​

Delayed actions​

Combine Kafka and RPC​

K/V State​

Event Enrichment​

Delayed actions​

Combine Kafka and RPC​

K/V State​

Event Enrichment​

Delayed actions​

Combine Kafka and RPC​

K/V State​

Event Enrichment​

Delayed actions​

Combine Kafka and RPC​

K/V State​

Event Enrichment​

Delayed actions​

Combine Kafka and RPC​

K/V State​

Event Enrichment​

Delayed actions​

Event processing with Restate

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Lightweight Durable Functions

Push events to functions

Queue per key

Postpone processing

Durable side effects

Flexible control flow

Stateful event processing with Restate

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment

Delayed actions

Combine Kafka and RPC

K/V State

Event Enrichment