add default for pekko.remote.akka.version (apache#1112)

* add default for pekko.cluster.akka.version

* refactor configs

* Update reference.conf

* add validations for config settings

* Update RemoteSettings.scala

* Update RemoteSettingsSpec.scala

* scalafmt

cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala

@@ -58,7 +58,7 @@ private[cluster] abstract class SeedNodeProcess(joinConfigCompatChecker: JoinCon
     if (cfg.hasPath("akka.version")) {
       cfg.getString("akka.version")
     } else {
-      cfg.getString("pekko.cluster.akka.version")
+      cfg.getString("pekko.remote.akka.version")
     }
   }
 

remote/src/main/resources/reference.conf

@@ -176,19 +176,32 @@ pekko {
     warn-unsafe-watch-outside-cluster = on
 
     # When receiving requests from other remote actors, what are the valid
-    # prefix's to check against. Useful for when dealing with rolling cluster
+    # prefixes to check against. Useful for when dealing with rolling cluster
     # migrations with compatible systems such as Lightbend's Akka.
-    accept-protocol-names = ["pekko", "akka"]
+    # By default, we only support "pekko" protocol.
+    # If you want to also support Akka, change this config to:
+    # pekko.remote.accept-protocol-names = ["pekko", "akka"]
+    # A ConfigurationException will be thrown at runtime if the array is empty
+    # or contains values other than "pekko" and/or "akka".
+    accept-protocol-names = ["pekko"]
 
     # The protocol name to use when sending requests to other remote actors.
     # Useful when dealing with rolling migration, i.e. temporarily change
     # the protocol name to match another compatible actor implementation
     # such as Lightbend's "akka" (whilst making sure accept-protocol-names
     # contains "akka") so that you can gracefully migrate all nodes to Apache
     # Pekko and then change the protocol-name back to "pekko" once all
-    # nodes have been are running on Apache Pekko
+    # nodes have been are running on Apache Pekko.
+    # A ConfigurationException will be thrown at runtime if the value is not
+    # set to "pekko" or "akka".
     protocol-name = "pekko"
 
+    # When pekko.remote.accept-protocol-names contains "akka", then we
+    # need to know the Akka version. If you include the Akka jars on the classpath,
+    # we can use the akka.version from their configuration. This configuration
+    # setting is only used if we can't find an akka.version setting.
+    akka.version = "2.6.21"
+
     # Settings for the Phi accrual failure detector (http://www.jaist.ac.jp/~defago/files/pdf/IS_RR_2004_010.pdf
     # [Hayashibara et al]) used for remote death watch.
     # The default PhiAccrualFailureDetector will trigger if there are no heartbeats within

remote/src/main/scala/org/apache/pekko/remote/RemoteSettings.scala

@@ -199,11 +199,30 @@ final class RemoteSettings(val config: Config) {
   @deprecated("Classic remoting is deprecated, use Artery", "Akka 2.6.0")
   val Adapters: Map[String, String] = configToMap(getConfig("pekko.remote.classic.adapters"))
 
-  val ProtocolName: String = getString("pekko.remote.protocol-name")
+  private val AllowableProtocolNames = Set("pekko", "akka")
 
-  val AcceptProtocolNames: Set[String] =
-    immutableSeq(getStringList("pekko.remote.accept-protocol-names")).toSet.requiring(_.nonEmpty,
-      "accept-protocol-names must be non empty")
+  val ProtocolName: String = {
+    val setting = getString("pekko.remote.protocol-name")
+    if (!AllowableProtocolNames.contains(setting)) {
+      throw new ConfigurationException("The only allowed values for pekko.remote.protocol-name " +
+        "are \"pekko\" and \"akka\".")
+    }
+    setting
+  }
+
+  val AcceptProtocolNames: Set[String] = {
+    val set = immutableSeq(getStringList("pekko.remote.accept-protocol-names")).toSet
+    if (set.isEmpty) {
+      throw new ConfigurationException("pekko.remote.accept-protocol-names setting must not be empty. " +
+        "The setting is an array and the only acceptable values are \"pekko\" and \"akka\".")
+    }
+    val filteredSet = set.filterNot(AllowableProtocolNames.contains)
+    if (filteredSet.nonEmpty) {
+      throw new ConfigurationException("pekko.remote.accept-protocol-names is an array setting " +
+        "that only accepts the values \"pekko\" and \"akka\".")
+    }
+    set
+  }
 
   private def transportNames: immutable.Seq[String] =
     immutableSeq(getStringList("pekko.remote.classic.enabled-transports"))

remote/src/test/scala/org/apache/pekko/remote/classic/RemoteSettingsSpec.scala

@@ -18,7 +18,10 @@ import com.typesafe.config.ConfigFactory
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
-import org.apache.pekko.remote.RemoteSettings
+import org.apache.pekko
+import pekko.ConfigurationException
+import pekko.remote.RemoteSettings
+import pekko.testkit.PekkoSpec
 
 @nowarn("msg=deprecated")
 class RemoteSettingsSpec extends AnyWordSpec with Matchers {
@@ -34,6 +37,32 @@ class RemoteSettingsSpec extends AnyWordSpec with Matchers {
           .parseString("pekko.remote.classic.log-frame-size-exceeding = 100b")
           .withFallback(ConfigFactory.load())).LogFrameSizeExceeding shouldEqual Some(100)
     }
+    "fail if unknown protocol name is used" in {
+      val cfg = ConfigFactory.parseString("pekko.remote.protocol-name=unknown")
+        .withFallback(PekkoSpec.testConf)
+      val ex = intercept[ConfigurationException] {
+        new RemoteSettings(ConfigFactory.load(cfg))
+      }
+      ex.getMessage shouldEqual
+      """The only allowed values for pekko.remote.protocol-name are "pekko" and "akka"."""
+    }
+    "fail if empty accept-protocol-names is used" in {
+      val cfg = ConfigFactory.parseString("pekko.remote.accept-protocol-names=[]")
+        .withFallback(PekkoSpec.testConf)
+      val ex = intercept[ConfigurationException] {
+        new RemoteSettings(ConfigFactory.load(cfg))
+      }
+      ex.getMessage should startWith("pekko.remote.accept-protocol-names setting must not be empty")
+    }
+    "fail if invalid accept-protocol-names value is used" in {
+      val cfg = ConfigFactory.parseString("""pekko.remote.accept-protocol-names=["pekko", "unknown"]""")
+        .withFallback(PekkoSpec.testConf)
+      val ex = intercept[ConfigurationException] {
+        new RemoteSettings(ConfigFactory.load(cfg))
+      }
+      ex.getMessage shouldEqual
+      """pekko.remote.accept-protocol-names is an array setting that only accepts the values "pekko" and "akka"."""
+    }
   }
 
 }