Execute no SQL when all events are rejected

Author: okumin

core/src/main/scala/akka/persistence/journal/sqlasync/ScalikeJDBCWriteJournal.scala

@@ -3,12 +3,11 @@ package akka.persistence.journal.sqlasync
 import akka.persistence.common.StoragePlugin
 import akka.persistence.journal.AsyncWriteJournal
 import akka.persistence.{AtomicWrite, PersistentRepr}
-import scalikejdbc._
-import scalikejdbc.async._
-
 import scala.collection.immutable
 import scala.concurrent.Future
-import scala.util.{Success, Try}
+import scala.util.{Failure, Success, Try}
+import scalikejdbc._
+import scalikejdbc.async._
 
 private[sqlasync] trait ScalikeJDBCWriteJournal extends AsyncWriteJournal with StoragePlugin {
   private[this] lazy val journalTable = {
@@ -51,8 +50,19 @@ private[sqlasync] trait ScalikeJDBCWriteJournal extends AsyncWriteJournal with S
           }
           (batch, result) = serialize(keys)
           sql = sql"INSERT INTO $journalTable (persistence_key, sequence_nr, message) VALUES $batch"
-          _ <- logging(sql).update().future()
-        } yield result
+          _ <- if (result.forall(_.isFailure)) {
+            // No insertion is needed.
+            Future.successful(())
+          } else {
+            logging(sql).update().future()
+          }
+        } yield {
+          result.foreach {
+            case Success(_) =>
+            case Failure(e) => log.warning("Failed serializing an event.", e)
+          }
+          result
+        }
       }
     }
   }

core/src/test/scala/akka/persistence/journal/sqlasync/MySQLAsyncJournalSpec.scala

@@ -1,9 +1,89 @@
 package akka.persistence.journal.sqlasync
 
+import akka.actor.Actor
+import akka.persistence.JournalProtocol.{WriteMessageRejected, WriteMessageSuccess, WriteMessages, WriteMessagesSuccessful}
 import akka.persistence.helper.MySQLInitializer
 import akka.persistence.journal.JournalSpec
+import akka.persistence.{AtomicWrite, PersistentImpl, PersistentRepr}
+import akka.testkit.TestProbe
 import com.typesafe.config.ConfigFactory
+import java.io.NotSerializableException
+import scala.concurrent.duration._
 
 class MySQLAsyncJournalSpec
   extends JournalSpec(ConfigFactory.load("mysql-application.conf"))
-  with MySQLInitializer
+  with MySQLInitializer {
+
+  "ScalikeJDBCWriteJournal" must {
+    "not execute SQL when all the events is not serializable" in {
+      val probe = TestProbe()
+
+      val notSerializableEvent = new Object { override def toString = "not serializable" }
+      val messages = (6 to 8).map { i =>
+        AtomicWrite(PersistentRepr(
+          payload = notSerializableEvent,
+          sequenceNr = i,
+          persistenceId = pid,
+          sender = Actor.noSender,
+          writerUuid = writerUuid
+        ))
+      }
+      journal ! WriteMessages(messages, probe.ref, actorInstanceId)
+      probe.expectMsg(WriteMessagesSuccessful)
+
+      val Pid = pid
+      val WriterUuid = writerUuid
+      probe.expectMsgPF() {
+        case WriteMessageRejected(PersistentImpl(payload, 6L, Pid, _, _, Actor.noSender, WriterUuid), cause, _) =>
+          payload should be(notSerializableEvent)
+          cause.isInstanceOf[NotSerializableException] should be(true)
+      }
+      probe.expectMsgPF() {
+        case WriteMessageRejected(PersistentImpl(payload, 7L, Pid, _, _, Actor.noSender, WriterUuid), cause, _) =>
+          payload should be(notSerializableEvent)
+          cause.isInstanceOf[NotSerializableException] should be(true)
+      }
+      probe.expectMsgPF() {
+        case WriteMessageRejected(PersistentImpl(payload, 8L, Pid, _, _, Actor.noSender, WriterUuid), cause, _) =>
+          payload should be(notSerializableEvent)
+          cause.isInstanceOf[NotSerializableException] should be(true)
+      }
+      probe.expectNoMsg(1.second)
+    }
+
+    "handle partial serialization errors" in {
+      val probe = TestProbe()
+
+      val notSerializableEvent = new Object { override def toString = "not serializable" }
+      val messages = (6 to 8).map { i =>
+        val event = if (i == 7) notSerializableEvent else s"b-$i"
+        AtomicWrite(PersistentRepr(
+          payload = event,
+          sequenceNr = i,
+          persistenceId = pid,
+          sender = Actor.noSender,
+          writerUuid = writerUuid
+        ))
+      }
+      journal ! WriteMessages(messages, probe.ref, actorInstanceId)
+      probe.expectMsg(WriteMessagesSuccessful)
+
+      val Pid = pid
+      val WriterUuid = writerUuid
+      probe.expectMsgPF() {
+        case WriteMessageSuccess(PersistentImpl(payload, 6L, Pid, _, _, Actor.noSender, WriterUuid), _) =>
+          payload should be("b-6")
+      }
+      probe.expectMsgPF() {
+        case WriteMessageRejected(PersistentImpl(payload, 7L, Pid, _, _, Actor.noSender, WriterUuid), cause, _) =>
+          payload should be(notSerializableEvent)
+          cause.isInstanceOf[NotSerializableException] should be(true)
+      }
+      probe.expectMsgPF() {
+        case WriteMessageSuccess(PersistentImpl(payload, 8L, Pid, _, _, Actor.noSender, WriterUuid), _) =>
+          payload should be("b-8")
+      }
+      probe.expectNoMsg(1.second)
+    }
+  }
+}