Log failed lease releases when a Shard stops #3705

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)
+      }
+    }
+
 }

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(_) => () }
+        }
+    }
   }
 
 }

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