Packages

final class KafkaClusterSharding extends Extension

API MAY CHANGE

Akka Extension to enable Akka Cluster External Sharding with Alpakka Kafka.

Annotations
@ApiMayChange()
Source
KafkaClusterSharding.scala
Linear Supertypes
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. KafkaClusterSharding
  2. Extension
  3. AnyRef
  4. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. Protected

Instance Constructors

  1. new KafkaClusterSharding(system: ExtendedActorSystem)

Value Members

  1. def messageExtractor[M](kafkaPartitions: Int): KafkaShardingMessageExtractor[M]

    API MAY CHANGE

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy is provided explicitly with kafkaPartitions.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  2. def messageExtractor[M](topic: String, timeout: Duration, settings: ConsumerSettings[_, _]): CompletionStage[KafkaShardingMessageExtractor[M]]

    Java API

    Java API

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka cluster for the number of partitions of a user provided topic. Use the settings parameter to configure the Kafka Consumer connection required to retrieve the number of partitions. Each call to this method will result in a round trip to Kafka. This method should only be called once per entity type M, per local actor system.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  3. def messageExtractor[M](topic: String, timeout: FiniteDuration, settings: ConsumerSettings[_, _]): Future[KafkaShardingMessageExtractor[M]]

    API MAY CHANGE

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka cluster for the number of partitions of a user provided topic. Use the settings parameter to configure the Kafka Consumer connection required to retrieve the number of partitions. Each call to this method will result in a round trip to Kafka. This method should only be called once per entity type M, per local actor system.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  4. def messageExtractorNoEnvelope[M](kafkaPartitions: Int, entityIdExtractor: Function[M, String]): KafkaShardingNoEnvelopeExtractor[M]

    API MAY CHANGE

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy is provided explicitly with kafkaPartitions.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  5. def messageExtractorNoEnvelope[M](kafkaPartitions: Int, entityIdExtractor: (M) => String): KafkaShardingNoEnvelopeExtractor[M]

    API MAY CHANGE

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy is provided explicitly with kafkaPartitions.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  6. def messageExtractorNoEnvelope[M](topic: String, timeout: Duration, entityIdExtractor: Function[M, String], settings: ConsumerSettings[_, _]): CompletionStage[KafkaShardingNoEnvelopeExtractor[M]]

    Java API

    Java API

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka cluster for the number of partitions of a user provided topic. Use the settings parameter to configure the Kafka Consumer connection required to retrieve the number of partitions. Use the entityIdExtractor to pick a field from the Entity to use as the entity id for the hashing strategy. Each call to this method will result in a round trip to Kafka. This method should only be called once per entity type M, per local actor system.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  7. def messageExtractorNoEnvelope[M](topic: String, timeout: FiniteDuration, entityIdExtractor: (M) => String, settings: ConsumerSettings[_, _]): Future[KafkaShardingNoEnvelopeExtractor[M]]

    API MAY CHANGE

    API MAY CHANGE

    Asynchronously return a akka.cluster.sharding.typed.ShardingMessageExtractor with a default hashing strategy based on Apache Kafka's org.apache.kafka.clients.producer.internals.DefaultPartitioner.

    The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka cluster for the number of partitions of a user provided topic. Use the settings parameter to configure the Kafka Consumer connection required to retrieve the number of partitions. Use the entityIdExtractor to pick a field from the Entity to use as the entity id for the hashing strategy. Each call to this method will result in a round trip to Kafka. This method should only be called once per entity type M, per local actor system.

    All topics used in a Consumer akka.kafka.Subscription must contain the same number of partitions to ensure that entities are routed to the same Entity type.

    Annotations
    @ApiMayChange()
  8. def rebalanceListener(typeKey: EntityTypeKey[_]): ActorRef[ConsumerRebalanceEvent]

    Java API

    Java API

    API MAY CHANGE

    Create an Alpakka Kafka rebalance listener that handles TopicPartitionsAssigned events. The typeKey is used to create the ExternalShardAllocation client. When partitions are assigned to this consumer group member the rebalance listener will use the ExternalShardAllocation client to update the External Sharding strategy accordingly so that entities are (eventually) routed to the local Akka cluster member.

    Returns an Akka typed akka.actor.typed.ActorRef. This must be converted to a classic actor before it can be passed to an Alpakka Kafka ConsumerSettings.

    import akka.actor.typed.scaladsl.adapter._
    val listenerClassicActorRef: akka.actor.ActorRef = listenerTypedActorRef.toClassic
    Annotations
    @ApiMayChange()
  9. def rebalanceListener(typeKey: EntityTypeKey[_]): ActorRef[ConsumerRebalanceEvent]

    API MAY CHANGE

    API MAY CHANGE

    Create an Alpakka Kafka rebalance listener that handles TopicPartitionsAssigned events. The typeKey is used to create the ExternalShardAllocation client. When partitions are assigned to this consumer group member the rebalance listener will use the ExternalShardAllocation client to update the External Sharding strategy accordingly so that entities are (eventually) routed to the local Akka cluster member.

    Returns an Akka typed akka.actor.typed.ActorRef. This must be converted to a classic actor before it can be passed to an Alpakka Kafka ConsumerSettings.

    import akka.actor.typed.scaladsl.adapter._
    val listenerClassicActorRef: akka.actor.ActorRef = listenerTypedActorRef.toClassic
    Annotations
    @ApiMayChange()