diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala index 27ed10a2de..0da1df79c0 100644 --- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala +++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala @@ -18,7 +18,11 @@ import java.nio.charset.StandardCharsets import java.util import scala.annotation.nowarn +import scala.concurrent.Future import scala.concurrent.duration._ +import scala.util.Failure +import scala.util.Success +import scala.util.Try import org.apache.pekko import pekko.actor.Actor @@ -1235,8 +1239,30 @@ private[pekko] class Shard( override def postStop(): Unit = { passivateIntervalTask.foreach(_.cancel()) - lease.foreach(_.release()) + releaseLease() log.debug("{}: Shard [{}] shutting down", typeName, shardId) } + private def releaseLease(): Unit = + lease.foreach { l => + // The shard is stopping, so log from the callback; the adapter is thread-safe. + val releaseLog = log + val tn = typeName + val id = shardId + Future.fromTry(Try(l.release())).flatten.onComplete { + case Success(true) => + case Success(false) => + releaseLog.error( + "{}: Failed to release lease for shard id [{}]. Shard may not be able to start elsewhere until lease timeout", + tn, + id) + case Failure(t) => + releaseLog.error( + t, + "{}: Failed to release lease for shard id [{}]. Shard may not be able to start elsewhere until lease timeout", + tn, + id) + } + } + } diff --git a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala index ae35544762..8799c77e3b 100644 --- a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala +++ b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala @@ -25,7 +25,7 @@ import pekko.actor.Props import pekko.cluster.Cluster import pekko.cluster.MemberStatus import pekko.cluster.sharding.ShardRegion.ShardId -import pekko.coordination.lease.{ LeaseUsageSettings, TestLeaseExt } +import pekko.coordination.lease.{ LeaseUsageSettings, TestLease, TestLeaseExt } import pekko.testkit.EventFilter import pekko.testkit.PekkoSpec import pekko.testkit.TestProbe @@ -162,6 +162,18 @@ class ShardWithLeaseSpec extends PekkoSpec(ShardWithLeaseSpec.config) with WithL sharding.tell(EntityEnvelope(1, "hello"), probe.ref) probe.expectNoMessage(shortDuration) } + + "log an error if releasing the lease returns false when the shard stops" in new Setup { + expectReleaseFailureLoggedOnStop(Future.successful(false)) + } + + "log an error if releasing the lease fails when the shard stops" in new Setup { + expectReleaseFailureLoggedOnStop(Future.failed(BadLease("release failed"))) + } + + "log an error if releasing the lease throws when the shard stops" in new Setup { + expectReleaseFailureLoggedOnStop(throw BadLease("release threw")) + } } var typeIdx = 0 @@ -187,6 +199,23 @@ class ShardWithLeaseSpec extends PekkoSpec(ShardWithLeaseSpec.config) with WithL else s"$customLeaseName-$shardId" testLeaseExt.getTestLease(leaseName) } + + def expectReleaseFailureLoggedOnStop(releaseResult: => Future[Boolean]): Unit = { + val probe = TestProbe() + sharding.tell(EntityEnvelope(1, "hello"), probe.ref) + val lease = leaseFor("1") + lease.initialPromise.complete(Success(true)) + probe.expectMsg("ack hello") + + lease.setNextReleaseResult(releaseResult) + EventFilter + .error(start = s"$typeName: Failed to release lease for shard id [1]", occurrences = 1) + .intercept { + // losing the lease stops the shard, which releases the lease in postStop + lease.getCurrentCallback().apply(Some(BadLease("bye bye lease"))) + lease.probe.fishForSpecificMessage() { case TestLease.ReleaseReq(_) => () } + } + } } } diff --git a/coordination/src/test/scala/org/apache/pekko/coordination/lease/TestLease.scala b/coordination/src/test/scala/org/apache/pekko/coordination/lease/TestLease.scala index 71b218b696..9c028d69b8 100644 --- a/coordination/src/test/scala/org/apache/pekko/coordination/lease/TestLease.scala +++ b/coordination/src/test/scala/org/apache/pekko/coordination/lease/TestLease.scala @@ -77,6 +77,7 @@ class TestLease(settings: LeaseSettings, system: ExtendedActorSystem) extends Le val initialPromise = Promise[Boolean]() private val nextAcquireResult = new AtomicReference[Future[Boolean]](initialPromise.future) + private val nextReleaseResult = new AtomicReference[() => Future[Boolean]](() => Future.successful(true)) private val nextCheckLeaseResult = new AtomicReference[Boolean](false) private val currentCallBack = new AtomicReference[Option[Throwable] => Unit](_ => ()) @@ -87,6 +88,10 @@ class TestLease(settings: LeaseSettings, system: ExtendedActorSystem) extends Le def setNextAcquireResult(next: Future[Boolean]): Unit = nextAcquireResult.set(next) + // Evaluate on the next release only, so tests can also simulate synchronous failures. + def setNextReleaseResult(next: => Future[Boolean]): Unit = + nextReleaseResult.set(() => next) + def setNextCheckLeaseResult(value: Boolean): Unit = nextCheckLeaseResult.set(value) @@ -100,7 +105,7 @@ class TestLease(settings: LeaseSettings, system: ExtendedActorSystem) extends Le override def release(): Future[Boolean] = { probe.ref ! ReleaseReq(settings.ownerName) - Future.successful(true) + nextReleaseResult.getAndSet(() => Future.successful(true))() } override def checkLease(): Boolean = nextCheckLeaseResult.get