Move Lease usage settings inside akka-coordination (#26637)
This commit is contained in:
parent
791e203ca4
commit
a39ac61265
4 changed files with 19 additions and 16 deletions
|
|
@ -10,8 +10,9 @@ import akka.actor.ActorSystem
|
||||||
import akka.actor.NoSerializationVerificationNeeded
|
import akka.actor.NoSerializationVerificationNeeded
|
||||||
import akka.annotation.{ ApiMayChange, InternalApi }
|
import akka.annotation.{ ApiMayChange, InternalApi }
|
||||||
import com.typesafe.config.Config
|
import com.typesafe.config.Config
|
||||||
import akka.cluster.{ Cluster, ClusterLeaseSettings }
|
import akka.cluster.Cluster
|
||||||
import akka.cluster.singleton.ClusterSingletonManagerSettings
|
import akka.cluster.singleton.ClusterSingletonManagerSettings
|
||||||
|
import akka.coordination.lease.LeaseUsageSettings
|
||||||
import akka.util.JavaDurationConverters._
|
import akka.util.JavaDurationConverters._
|
||||||
|
|
||||||
object ClusterShardingSettings {
|
object ClusterShardingSettings {
|
||||||
|
|
@ -61,7 +62,7 @@ object ClusterShardingSettings {
|
||||||
|
|
||||||
val lease = config.getString("use-lease") match {
|
val lease = config.getString("use-lease") match {
|
||||||
case s if s.isEmpty ⇒ None
|
case s if s.isEmpty ⇒ None
|
||||||
case other ⇒ Some(new ClusterLeaseSettings(other, config.getDuration("lease-retry-interval").asScala))
|
case other ⇒ Some(new LeaseUsageSettings(other, config.getDuration("lease-retry-interval").asScala))
|
||||||
}
|
}
|
||||||
|
|
||||||
new ClusterShardingSettings(
|
new ClusterShardingSettings(
|
||||||
|
|
@ -220,7 +221,7 @@ final class ClusterShardingSettings(
|
||||||
val passivateIdleEntityAfter: FiniteDuration,
|
val passivateIdleEntityAfter: FiniteDuration,
|
||||||
val tuningParameters: ClusterShardingSettings.TuningParameters,
|
val tuningParameters: ClusterShardingSettings.TuningParameters,
|
||||||
val coordinatorSingletonSettings: ClusterSingletonManagerSettings,
|
val coordinatorSingletonSettings: ClusterSingletonManagerSettings,
|
||||||
val leaseSettings: Option[ClusterLeaseSettings])
|
val leaseSettings: Option[LeaseUsageSettings])
|
||||||
extends NoSerializationVerificationNeeded {
|
extends NoSerializationVerificationNeeded {
|
||||||
|
|
||||||
// bin compat for 2.5.21
|
// bin compat for 2.5.21
|
||||||
|
|
@ -302,7 +303,7 @@ final class ClusterShardingSettings(
|
||||||
copy(passivateIdleAfter = duration.asScala)
|
copy(passivateIdleAfter = duration.asScala)
|
||||||
|
|
||||||
@ApiMayChange
|
@ApiMayChange
|
||||||
def withLeaseSettings(leaseSettings: ClusterLeaseSettings): ClusterShardingSettings =
|
def withLeaseSettings(leaseSettings: LeaseUsageSettings): ClusterShardingSettings =
|
||||||
copy(leaseSettings = Some(leaseSettings))
|
copy(leaseSettings = Some(leaseSettings))
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -322,7 +323,7 @@ final class ClusterShardingSettings(
|
||||||
passivateIdleAfter: FiniteDuration = passivateIdleEntityAfter,
|
passivateIdleAfter: FiniteDuration = passivateIdleEntityAfter,
|
||||||
tuningParameters: ClusterShardingSettings.TuningParameters = tuningParameters,
|
tuningParameters: ClusterShardingSettings.TuningParameters = tuningParameters,
|
||||||
coordinatorSingletonSettings: ClusterSingletonManagerSettings = coordinatorSingletonSettings,
|
coordinatorSingletonSettings: ClusterSingletonManagerSettings = coordinatorSingletonSettings,
|
||||||
leaseSettings: Option[ClusterLeaseSettings] = leaseSettings): ClusterShardingSettings =
|
leaseSettings: Option[LeaseUsageSettings] = leaseSettings): ClusterShardingSettings =
|
||||||
new ClusterShardingSettings(
|
new ClusterShardingSettings(
|
||||||
role,
|
role,
|
||||||
rememberEntities,
|
rememberEntities,
|
||||||
|
|
|
||||||
|
|
@ -7,8 +7,9 @@ package akka.cluster.sharding
|
||||||
import java.util.concurrent.atomic.AtomicInteger
|
import java.util.concurrent.atomic.AtomicInteger
|
||||||
|
|
||||||
import akka.actor.{ Actor, ActorLogging, PoisonPill, Props }
|
import akka.actor.{ Actor, ActorLogging, PoisonPill, Props }
|
||||||
import akka.cluster.{ ClusterLeaseSettings, TestLeaseExt }
|
import akka.cluster.TestLeaseExt
|
||||||
import akka.cluster.sharding.ShardRegion.ShardInitialized
|
import akka.cluster.sharding.ShardRegion.ShardInitialized
|
||||||
|
import akka.coordination.lease.LeaseUsageSettings
|
||||||
import akka.testkit.{ AkkaSpec, ImplicitSender, TestProbe }
|
import akka.testkit.{ AkkaSpec, ImplicitSender, TestProbe }
|
||||||
|
|
||||||
import scala.concurrent.Future
|
import scala.concurrent.Future
|
||||||
|
|
@ -100,7 +101,7 @@ class ShardSpec extends AkkaSpec(ShardSpec.config) with ImplicitSender {
|
||||||
trait Setup {
|
trait Setup {
|
||||||
val shardId = nextShardId
|
val shardId = nextShardId
|
||||||
val parent = TestProbe()
|
val parent = TestProbe()
|
||||||
val settings = ClusterShardingSettings(system).withLeaseSettings(new ClusterLeaseSettings("test-lease", 2.seconds))
|
val settings = ClusterShardingSettings(system).withLeaseSettings(new LeaseUsageSettings("test-lease", 2.seconds))
|
||||||
def lease = awaitAssert {
|
def lease = awaitAssert {
|
||||||
testLeaseExt.getTestLease(leaseNameForShard(typeName, shardId))
|
testLeaseExt.getTestLease(leaseNameForShard(typeName, shardId))
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,7 @@ import scala.concurrent.Promise
|
||||||
import akka.Done
|
import akka.Done
|
||||||
import akka.actor.CoordinatedShutdown
|
import akka.actor.CoordinatedShutdown
|
||||||
import akka.annotation.{ ApiMayChange, DoNotInherit }
|
import akka.annotation.{ ApiMayChange, DoNotInherit }
|
||||||
|
import akka.coordination.lease.LeaseUsageSettings
|
||||||
import akka.pattern.ask
|
import akka.pattern.ask
|
||||||
import akka.util.Timeout
|
import akka.util.Timeout
|
||||||
import akka.coordination.lease.scaladsl.{ Lease, LeaseProvider }
|
import akka.coordination.lease.scaladsl.{ Lease, LeaseProvider }
|
||||||
|
|
@ -56,7 +57,7 @@ object ClusterSingletonManagerSettings {
|
||||||
val lease = config.getString("use-lease") match {
|
val lease = config.getString("use-lease") match {
|
||||||
case s if s.isEmpty ⇒ None
|
case s if s.isEmpty ⇒ None
|
||||||
case leaseConfigPath =>
|
case leaseConfigPath =>
|
||||||
Some(new ClusterLeaseSettings(leaseConfigPath, config.getDuration("lease-retry-interval").asScala))
|
Some(new LeaseUsageSettings(leaseConfigPath, config.getDuration("lease-retry-interval").asScala))
|
||||||
}
|
}
|
||||||
new ClusterSingletonManagerSettings(
|
new ClusterSingletonManagerSettings(
|
||||||
singletonName = config.getString("singleton-name"),
|
singletonName = config.getString("singleton-name"),
|
||||||
|
|
@ -113,7 +114,7 @@ final class ClusterSingletonManagerSettings(
|
||||||
val role: Option[String],
|
val role: Option[String],
|
||||||
val removalMargin: FiniteDuration,
|
val removalMargin: FiniteDuration,
|
||||||
val handOverRetryInterval: FiniteDuration,
|
val handOverRetryInterval: FiniteDuration,
|
||||||
val leaseSettings: Option[ClusterLeaseSettings])
|
val leaseSettings: Option[LeaseUsageSettings])
|
||||||
extends NoSerializationVerificationNeeded {
|
extends NoSerializationVerificationNeeded {
|
||||||
|
|
||||||
// bin compat for akka 2.5.21
|
// bin compat for akka 2.5.21
|
||||||
|
|
@ -137,7 +138,7 @@ final class ClusterSingletonManagerSettings(
|
||||||
def withHandOverRetryInterval(retryInterval: FiniteDuration): ClusterSingletonManagerSettings =
|
def withHandOverRetryInterval(retryInterval: FiniteDuration): ClusterSingletonManagerSettings =
|
||||||
copy(handOverRetryInterval = retryInterval)
|
copy(handOverRetryInterval = retryInterval)
|
||||||
|
|
||||||
def withLeaseSettings(leaseSettings: ClusterLeaseSettings): ClusterSingletonManagerSettings =
|
def withLeaseSettings(leaseSettings: LeaseUsageSettings): ClusterSingletonManagerSettings =
|
||||||
copy(leaseSettings = Some(leaseSettings))
|
copy(leaseSettings = Some(leaseSettings))
|
||||||
|
|
||||||
private def copy(
|
private def copy(
|
||||||
|
|
@ -145,7 +146,7 @@ final class ClusterSingletonManagerSettings(
|
||||||
role: Option[String] = role,
|
role: Option[String] = role,
|
||||||
removalMargin: FiniteDuration = removalMargin,
|
removalMargin: FiniteDuration = removalMargin,
|
||||||
handOverRetryInterval: FiniteDuration = handOverRetryInterval,
|
handOverRetryInterval: FiniteDuration = handOverRetryInterval,
|
||||||
leaseSettings: Option[ClusterLeaseSettings] = leaseSettings): ClusterSingletonManagerSettings =
|
leaseSettings: Option[LeaseUsageSettings] = leaseSettings): ClusterSingletonManagerSettings =
|
||||||
new ClusterSingletonManagerSettings(singletonName, role, removalMargin, handOverRetryInterval, leaseSettings)
|
new ClusterSingletonManagerSettings(singletonName, role, removalMargin, handOverRetryInterval, leaseSettings)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -2,17 +2,17 @@
|
||||||
* Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
|
* Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package akka.cluster
|
package akka.coordination.lease
|
||||||
|
|
||||||
import akka.annotation.ApiMayChange
|
import akka.annotation.ApiMayChange
|
||||||
|
|
||||||
import scala.concurrent.duration.FiniteDuration
|
|
||||||
import akka.util.JavaDurationConverters._
|
import akka.util.JavaDurationConverters._
|
||||||
import akka.util.PrettyDuration._
|
import akka.util.PrettyDuration._
|
||||||
|
|
||||||
|
import scala.concurrent.duration.FiniteDuration
|
||||||
|
|
||||||
@ApiMayChange
|
@ApiMayChange
|
||||||
class ClusterLeaseSettings private[akka] (val leaseImplementation: String, val leaseRetryInterval: FiniteDuration) {
|
class LeaseUsageSettings private[akka] (val leaseImplementation: String, val leaseRetryInterval: FiniteDuration) {
|
||||||
def getLeaseRetryInterval(): java.time.Duration = leaseRetryInterval.asJava
|
def getLeaseRetryInterval(): java.time.Duration = leaseRetryInterval.asJava
|
||||||
|
|
||||||
override def toString = s"ClusterLeaseSettings($leaseImplementation, ${leaseRetryInterval.pretty})"
|
override def toString = s"LeaseUsageSettings($leaseImplementation, ${leaseRetryInterval.pretty})"
|
||||||
}
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue