Package akka.javasdk.consumer

Class Consumer

Object
akka.javasdk.consumer.Consumer
public abstract class Consumer extends Object
Consumers are stateless components that enable stream-based interaction between Akka services and other systems. They can be used to implement various use cases including:
  • Consume events from an Event Sourced Entity within the same service
  • Consume state changes from a Key Value Entity within the same service
  • Consume state changes from a Workflow within the same service
  • Consume events or state from an Entity or Workflow in another service using service-to-service eventing
  • Consume messages from Google Cloud Pub/Sub or Apache Kafka topics
  • Produce messages to Google Cloud Pub/Sub or Apache Kafka topics

Events and messages are guaranteed to be delivered at least once, which means Consumers must be able to handle duplicated messages.

A Consumer method should return an Consumer.Effect that describes what to do next. The Effect API defines the operations that Akka should perform when an incoming message is delivered to the Consumer.

Example usage: 


 @ComponentId("counter-events-consumer")
 @Consume.FromEventSourcedEntity(CounterEntity.class)
 public class CounterEventsConsumer extends Consumer {

   public Effect onEvent(CounterEvent event) {
     return switch (event) {
       case ValueIncreased valueIncreased ->
 //processing the event
 effects().done();
       case ValueMultiplied valueMultiplied -> effects().ignore();
     };
   }
 }

Concrete classes can accept the following types to the constructor:

Concrete classes must be annotated with ComponentId and one of the Consume annotations such as:

  • @Consume.FromEventSourcedEntity - to consume events from an Event Sourced Entity
  • @Consume.FromKeyValueEntity - to consume state changes from a Key Value Entity
  • @Consume.FromWorkflow - to consume state changes from a Workflow
  • @Consume.FromTopic - to consume messages from a message broker topic
  • @Consume.FromServiceStream - to consume events from another Akka service

For producing messages, use @Produce.ToTopic or @Produce.ServiceStream annotations.

If an exception is raised during message processing, the Akka runtime will redeliver the message until the application processes it without failures.

  • Constructor Details

    • Consumer

      public Consumer()

  • Method Details

    • messageContext

      protected final MessageContext messageContext()
      Additional context and metadata for a message handler.

      The message context provides access to:

      • Message metadata including CloudEvent attributes like subject id via messageContext().metadata().get("ce-subject")
      • CloudEvent interface for accessing standard CloudEvent properties
      • Information about message origin and region for multi-region deployments

      For Event Sourced and Key Value Entity consumers, the entity id is available through the metadata as the CloudEvent subject: messageContext().metadata().asCloudEvent().subject().get()

      This method will throw an exception if accessed from the constructor. It is only available when handling a message.

      Returns:
      the message context containing metadata and additional information about the current message
      Throws:
      IllegalStateException - if called outside of message handling (e.g., from constructor)

    • effects

      public final Consumer.Effect.Builder effects()
      Returns the Effect Builder for constructing effects that describe what the runtime should do after message processing.

      Use this to create effects such as:

      • effects().done() - to mark the message as successfully processed
      • effects().produce(message) - to produce a message to a topic or stream
      • effects().ignore() - to ignore the current message and continue processing
      Returns:
      the Effect Builder for creating Consumer effects

    • timers

      public final TimerScheduler timers()
      Returns a TimerScheduler that can be used to schedule actions to be executed at a later time.

      Timers can be used to schedule delayed processing, implement timeouts, or trigger periodic actions. The scheduled actions will be delivered as messages to the Consumer.

      This method can only be called when handling a message, not from the constructor.

      Returns:
      the TimerScheduler for scheduling delayed actions
      Throws:
      IllegalStateException - if called outside of message handling (e.g., from constructor)