fix: make AbstractPersistentActorWithTimers and AbstractFSMWithStash

actor-tests/src/test/java/org/apache/pekko/actor/AbstractFSMWithStashActorTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
+import org.apache.pekko.testkit.PekkoSpec;
+import org.apache.pekko.testkit.TestProbe;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.scalatestplus.junit.JUnitSuite;
+
+/**
+ * Subclassing {@link AbstractFSMWithStash} from Java has to keep compiling: both {@code FSM} and
+ * {@code UnrestrictedStash} implement {@code postStop}, and javac rejects a subclass unless the
+ * Scala base class carries a real (non-synthetic) override for it.
+ */
+public class AbstractFSMWithStashActorTest extends JUnitSuite {
+
+  public static class MyFSM extends AbstractFSMWithStash<String, String> {
+
+    private final ActorRef probe;
+
+    MyFSM(ActorRef probe) {
+      this.probe = probe;
+      startWith("start", "data");
+      when(
+          "start",
+          matchEvent(
+              String.class,
+              (event, data) -> {
+                if ("go".equals(event)) {
+                  unstashAll();
+                  return goTo("next");
+                } else {
+                  stash();
+                  return stay();
+                }
+              }));
+      when(
+          "next",
+          matchEvent(
+              String.class,
+              (event, data) -> {
+                probe.tell(event, getSelf());
+                return stay();
+              }));
+      initialize();
+    }
+  }
+
+  @ClassRule
+  public static PekkoJUnitActorSystemResource actorSystemResource =
+      new PekkoJUnitActorSystemResource("AbstractFSMWithStashActorTest", PekkoSpec.testConf());
+
+  private final ActorSystem system = actorSystemResource.getSystem();
+
+  @Test
+  public void canCreateFSMWithStash() {
+    TestProbe probe = new TestProbe(system);
+
+    ActorRef ref = system.actorOf(Props.create(MyFSM.class, probe.ref()));
+    ref.tell("work", ActorRef.noSender());
+    ref.tell("go", ActorRef.noSender());
+
+    probe.expectMsg("work");
+  }
+}

actor/src/main/scala/org/apache/pekko/actor/AbstractFSM.scala

@@ -572,4 +572,15 @@ abstract class AbstractLoggingFSM[S, D] extends AbstractFSM[S, D] with LoggingFS
  *
  * Finite State Machine actor abstract base class with Stash support.
  */
-abstract class AbstractFSMWithStash[S, D] extends AbstractFSM[S, D] with Stash
+abstract class AbstractFSMWithStash[S, D] extends AbstractFSM[S, D] with Stash {
+
+  // Overridden solely so that this class carries a real (non-synthetic) override of the member that
+  // both `FSM` and `UnrestrictedStash` implement. Scala 3 emits the mixin forwarder as an
+  // ACC_BRIDGE/ACC_SYNTHETIC method, which javac ignores when resolving inherited members, so a Java
+  // subclass would otherwise fail to compile with "inherits unrelated defaults". The body just
+  // delegates to `super`, which resolves exactly like the forwarder it replaces.
+
+  // No `@throws(classOf[Exception])` here: `FSM.postStop` declares no checked exceptions, so a wider
+  // throws clause would not be a valid override for javac.
+  override def postStop(): Unit = super.postStop()
+}

persistence/src/main/scala/org/apache/pekko/persistence/PersistentActor.scala

@@ -492,4 +492,20 @@ abstract class AbstractPersistentActor extends AbstractActor with AbstractPersis
 /**
  * Java API: Combination of [[AbstractPersistentActor]] and [[pekko.actor.AbstractActorWithTimers]].
  */
-abstract class AbstractPersistentActorWithTimers extends AbstractActor with Timers with AbstractPersistentActorLike
+abstract class AbstractPersistentActorWithTimers extends AbstractActor with Timers with AbstractPersistentActorLike {
+
+  // The methods below are overridden solely so that this class carries real (non-synthetic) overrides
+  // for the members that both `Timers` and `Eventsourced` implement. Scala 3 emits the mixin forwarders
+  // as ACC_BRIDGE/ACC_SYNTHETIC methods, which javac ignores when resolving inherited members, so a Java
+  // subclass would otherwise fail to compile with "inherits unrelated defaults". The bodies just delegate
+  // to `super`, which resolves exactly like the forwarders they replace.
+
+  override protected[pekko] def aroundReceive(receive: Actor.Receive, msg: Any): Unit =
+    super.aroundReceive(receive, msg)
+
+  override protected[pekko] def aroundPreRestart(reason: Throwable, message: Option[Any]): Unit =
+    super.aroundPreRestart(reason, message)
+
+  override protected[pekko] def aroundPostStop(): Unit =
+    super.aroundPostStop()
+}

persistence/src/test/java/org/apache/pekko/persistence/JavaTimerPersistentActor.java

@@ -0,0 +1,69 @@
+/*
+ * 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.persistence;
+
+import java.time.Duration;
+import org.apache.pekko.actor.ActorRef;
+
+/**
+ * Subclassing {@link AbstractPersistentActorWithTimers} from Java has to keep compiling: both
+ * {@code Timers} and {@code Eventsourced} implement the {@code aroundReceive}/{@code
+ * aroundPreRestart}/ {@code aroundPostStop} members, and javac rejects a subclass unless the Scala
+ * base class carries real (non-synthetic) overrides for them.
+ */
+@SuppressWarnings("unchecked")
+public class JavaTimerPersistentActor extends AbstractPersistentActorWithTimers {
+
+  public static final class Scheduled {
+    public final Object msg;
+    public final ActorRef replyTo;
+
+    public Scheduled(Object msg, ActorRef replyTo) {
+      this.msg = msg;
+      this.replyTo = replyTo;
+    }
+  }
+
+  private final String name;
+
+  public JavaTimerPersistentActor(String name) {
+    this.name = name;
+  }
+
+  @Override
+  public String persistenceId() {
+    return name;
+  }
+
+  @Override
+  public Receive createReceiveRecover() {
+    return receiveBuilder().matchAny(msg -> {}).build();
+  }
+
+  @Override
+  public Receive createReceive() {
+    return receiveBuilder()
+        .match(Scheduled.class, scheduled -> scheduled.replyTo.tell(scheduled.msg, getSelf()))
+        .matchAny(
+            msg -> {
+              timers().startSingleTimer("key", new Scheduled(msg, getSender()), Duration.ZERO);
+              persist(msg, evt -> {});
+            })
+        .build();
+  }
+}

persistence/src/test/java/org/apache/pekko/persistence/fsm/JavaSubclassCompilationCheck.java

@@ -0,0 +1,37 @@
+/*
+ * 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.persistence.fsm;
+
+/**
+ * Compile-only guard for Java-facing base classes that mix in more than one Scala trait.
+ *
+ * <p>See {@code org.apache.pekko.actor.JavaSubclassCompilationCheck} for why declaring the subclass
+ * is the whole test. {@link AbstractPersistentLoggingFSM} has no other Java subclass in the build.
+ *
+ * <p>The class under guard is deprecated, so the subclass is marked deprecated too: javac's test
+ * configuration runs with `-Werror`, and a use of deprecated API inside a deprecated element does
+ * not warn. This mirrors {@code AbstractPersistentFSMTest}.
+ */
+public final class JavaSubclassCompilationCheck {
+
+  private JavaSubclassCompilationCheck() {}
+
+  @Deprecated
+  abstract static class PersistentLoggingFSM
+      extends AbstractPersistentLoggingFSM<PersistentFSM.FSMState, String, String> {}
+}

persistence/src/test/scala/org/apache/pekko/persistence/TimerPersistentActorSpec.scala

@@ -113,6 +113,12 @@ class TimerPersistentActorSpec extends PersistenceSpec(ConfigFactory.parseString
       expectMsg("msg2")
     }
 
+    "not discard timer msg due to stashing for a Java subclass of AbstractPersistentActorWithTimers" in {
+      val pa = system.actorOf(Props(classOf[JavaTimerPersistentActor], "p4"))
+      pa ! "msg4"
+      expectMsg("msg4")
+    }
+
     "reject wrong order of traits, PersistentActor with Timer" in {
       if (TraitOrder.canBeChecked) {
         val pa = system.actorOf(Props[WrongOrder]())

stream-tests/src/test/java/org/apache/pekko/stream/stage/JavaSubclassCompilationCheck.java

@@ -0,0 +1,32 @@
+/*
+ * 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.stream.stage;
+
+/**
+ * Compile-only guard for Java-facing base classes that mix in more than one Scala trait.
+ *
+ * <p>See {@code org.apache.pekko.actor.JavaSubclassCompilationCheck} for why declaring the subclass
+ * is the whole test. {@link AbstractInOutHandler} mixes in both {@code InHandler} and {@code
+ * OutHandler} and has no other Java subclass in the build.
+ */
+public final class JavaSubclassCompilationCheck {
+
+  private JavaSubclassCompilationCheck() {}
+
+  abstract static class InOutHandler extends AbstractInOutHandler {}
+}