This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 005e242489 fix(actor-typed): discard stale ReceiveTimeout after
cancelReceiveTimeout to avoid NPE (#3086)
005e242489 is described below
commit 005e242489cc5fd46c9938795555224a2afac130
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Jun 19 10:16:59 2026 +0800
fix(actor-typed): discard stale ReceiveTimeout after cancelReceiveTimeout
to avoid NPE (#3086)
* fix(actor-typed): cancelReceiveTimeout 后丢弃 stale ReceiveTimeout 避免 NPE
Motivation:
cancelReceiveTimeout() 将 receiveTimeoutMsg 置 null,
但已通过 scheduleOnce 入队邮箱的 classic ReceiveTimeout 无法撤回;
stale 消息以 null 传入 typed 行为栈, 导致 InterceptorImpl.receive
中 msg.getClass 抛出 NullPointerException。
Modification:
ActorAdapter.aroundReceive 在转发 receiveTimeoutMsg 前做 null 检查,
stale timeout 直接丢弃; 新增 CancelReceiveTimeoutSpec 回归测试。
Result:
typed actor 在 cancelReceiveTimeout 与 stale ReceiveTimeout 竞态下
不再崩溃, 行为栈对调用方透明。
* style(actor-typed-tests): 修正 CancelReceiveTimeoutSpec 的 license header
为标准 Apache
---
.../actor/typed/CancelReceiveTimeoutSpec.scala | 106 +++++++++++++++++++++
.../typed/internal/adapter/ActorAdapter.scala | 6 +-
2 files changed, 111 insertions(+), 1 deletion(-)
diff --git
a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
new file mode 100644
index 0000000000..c342ba9cfd
--- /dev/null
+++
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.actor.typed
+
+import scala.concurrent.duration._
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.testkit.typed.scaladsl.TestProbe
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.adapter._
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class CancelReceiveTimeoutSpec extends ScalaTestWithActorTestKit with
AnyWordSpecLike with LogCapturing {
+
+ sealed trait Command
+ case object SetAndCancelTimeout extends Command
+ case object Timeout extends Command
+ case object Ping extends Command
+ case object Pong extends Event
+
+ sealed trait Event
+ case object Done extends Event
+
+ // A no-op interceptor that deepens the behavior stack so the message
+ // passes through InterceptorImpl.receive — the exact code path where
+ // msg.getClass throws NPE on a null message.
+ private def noopInterceptor[T: ClassTag] = new BehaviorInterceptor[T, T] {
+ override def aroundReceive(
+ ctx: TypedActorContext[T],
+ msg: T,
+ target: BehaviorInterceptor.ReceiveTarget[T]): Behavior[T] =
+ target(ctx, msg)
+ }
+
+ "A typed actor with receive timeout" must {
+
+ // Regression test for #3084: cancelReceiveTimeout() sets
receiveTimeoutMsg to
+ // null, but a classic ReceiveTimeout already enqueued in the mailbox
cannot be
+ // retracted. Before the fix the stale ReceiveTimeout was forwarded to the
typed
+ // behavior stack as a null message, causing NPE in
InterceptorImpl.receive.
+ "silently discard a stale ReceiveTimeout after cancelReceiveTimeout" in {
+ val probe = TestProbe[Event]()
+
+ def behavior: Behavior[Command] =
+ Behaviors.setup { context =>
+ // Wrap in an interceptor so the message traverses
InterceptorImpl.receive,
+ // the exact location where msg.getClass throws NPE on a null
message.
+ Behaviors.intercept[Command, Command](() =>
noopInterceptor[Command]) {
+ Behaviors.receiveMessage {
+ case SetAndCancelTimeout =>
+ // Set a very long timeout (won't actually fire), then
immediately
+ // cancel it. This leaves receiveTimeoutMsg as null.
+ context.setReceiveTimeout(1.hour, Timeout)
+ context.cancelReceiveTimeout()
+ probe.ref ! Done
+ Behaviors.same
+
+ case Timeout =>
+ // Should never reach here: the timeout was cancelled and we
send
+ // the classic ReceiveTimeout ourselves to simulate the stale
case.
+ Behaviors.unhandled
+
+ case Ping =>
+ probe.ref ! Pong
+ Behaviors.same
+ }
+ }
+ }
+
+ val ref = spawn(behavior)
+
+ // Step 1: actor sets and cancels the receive timeout →
receiveTimeoutMsg is now null
+ ref ! SetAndCancelTimeout
+ probe.expectMessage(Done)
+
+ // Step 2: simulate a stale classic ReceiveTimeout arriving in the
mailbox.
+ // This is what happens when the scheduler fires ReceiveTimeout before
+ // cancelReceiveTimeout() is processed but the actor dequeues them in the
+ // opposite order.
+ ref.toClassic ! pekko.actor.ReceiveTimeout
+
+ // Step 3: verify the actor is still alive and responsive (no NPE crash).
+ ref ! Ping
+ probe.expectMessage(Pong)
+ }
+ }
+}
diff --git
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
index 5b7b131b6c..7bee9f44f1 100644
---
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
+++
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
@@ -100,7 +100,11 @@ import pekko.util.OptionVal
} else Terminated(ActorRefAdapter(ref))
handleSignal(msg)
case classic.ReceiveTimeout =>
- handleMessage(ctx.receiveTimeoutMsg)
+ // cancelReceiveTimeout() sets receiveTimeoutMsg to null, but a
classic ReceiveTimeout
+ // that was already enqueued in the mailbox before the cancel cannot
be retracted.
+ // Discard the stale timeout to avoid passing null into the typed
behavior stack (#3084).
+ val timeoutMsg = ctx.receiveTimeoutMsg
+ if (timeoutMsg != null) handleMessage(timeoutMsg)
case wrapped: AdaptMessage[Any, T] @unchecked =>
withSafelyAdapted(() => wrapped.adapt()) {
case AdaptWithRegisteredMessageAdapter(msg) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
