Merge pull request #15324 from akka/wip-15131-snapshotstore-directory-check-master-ban

=per #15131 Make LocalSnapshotStore fail if it can't create target directory (Forward Port)

akka-persistence/src/main/scala/akka/persistence/snapshot/local/LocalSnapshotStore.scala

@@ -110,7 +110,11 @@ private[persistence] class LocalSnapshotStore extends SnapshotStore with ActorLo
     }.filter(md ⇒ criteria.matches(md) && !saving.contains(md)).toVector
 
   override def preStart() {
-    if (!snapshotDir.exists) snapshotDir.mkdirs()
+    if (!snapshotDir.isDirectory) {
+      if (!snapshotDir.mkdirs()) {
+        throw new IOException(s"Failed to create snapshot directory [${snapshotDir.getCanonicalPath}]")
+      }
+    }
     super.preStart()
   }
 

akka-persistence/src/test/scala/akka/persistence/SnapshotDirectoryFailureSpec.scala

@@ -0,0 +1,53 @@
+/**
+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.persistence
+
+import akka.testkit.{ ImplicitSender, EventFilter, TestEvent, AkkaSpec }
+import java.io.{ IOException, File }
+import akka.actor.{ ActorInitializationException, Props, ActorRef }
+
+object SnapshotDirectoryFailureSpec {
+  val inUseSnapshotPath = "target/inUseSnapshotPath"
+
+  class TestProcessor(name: String, probe: ActorRef) extends Processor {
+
+    override def processorId: String = name
+
+    override def preStart(): Unit = ()
+
+    def receive = {
+      case s: String               ⇒ saveSnapshot(s)
+      case SaveSnapshotSuccess(md) ⇒ probe ! md.sequenceNr
+      case other                   ⇒ probe ! other
+    }
+  }
+}
+
+class SnapshotDirectoryFailureSpec extends AkkaSpec(PersistenceSpec.config("leveldb", "SnapshotDirectoryFailureSpec", extraConfig = Some(
+  s"""
+    |akka.persistence.snapshot-store.local.dir = "${SnapshotDirectoryFailureSpec.inUseSnapshotPath}"
+  """.stripMargin))) with ImplicitSender {
+
+  import SnapshotDirectoryFailureSpec._
+
+  val file = new File(inUseSnapshotPath)
+
+  override protected def atStartup() {
+    if (!file.createNewFile()) throw new IOException(s"Failed to create test file [${file.getCanonicalFile}]")
+  }
+
+  override protected def afterTermination() {
+    if (!file.delete()) throw new IOException(s"Failed to delete test file [${file.getCanonicalFile}]")
+  }
+
+  "A local snapshot store configured with an failing directory name " must {
+    "throw an exception at startup" in {
+      EventFilter[ActorInitializationException](occurrences = 1).intercept {
+        val processor = system.actorOf(Props(classOf[TestProcessor], "SnapshotDirectoryFailureSpec-1", testActor))
+        processor ! "blahonga"
+      }
+    }
+  }
+}

akka-persistence/src/test/scala/akka/persistence/SnapshotSerializationSpec.scala

@@ -60,7 +60,6 @@ object SnapshotSerializationSpec {
 
 class SnapshotSerializationSpec extends AkkaSpec(PersistenceSpec.config("leveldb", "SnapshotSerializationSpec", serialization = "off", extraConfig = Some(
   """
-    |akka.logelevel = debug
     |akka.actor {
     |  serializers {
     |    my-snapshot = "akka.persistence.SnapshotSerializationSpec$MySerializer"