Update scalatest to 3.1.4

.circleci/config.yml

@@ -1,9 +1,6 @@
 version: 2
 jobs:
   build:
-    branches:
-      ignore:
-        - master
     environment:
       - JAVA_OPTS: "-XX:ReservedCodeCacheSize=256M -Xms1g -Xmx3g -Xss2m"
       - SBT_TEST_TIME_FACTOR: "2"

.gitignore

@@ -28,3 +28,4 @@ data/
 .DS_Store
 .credentials
 .gpgCredentials
+/.bsp/sbt.json

.scalafmt.conf

@@ -1,4 +1,4 @@
-version = 2.4.2
+version = 2.7.5
 style                           = defaultWithAlign
 danglingParentheses             = true
 indentOperator                  = spray

README.md

@@ -1,8 +1,8 @@
 # akka-persistence-kafka
 
-[![CircleCI](https://circleci.com/gh/j5ik2o/akka-persistence-kafka/tree/master.svg?style=shield&circle-token=c809688daf71f6ae582dd2d58cb5518401498373)](https://circleci.com/gh/j5ik2o/akka-persistence-kafka/tree/master)
-[![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.github.j5ik2o/akka-persistence-kafka_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.github.j5ik2o/akka-persistence-kafka_2.12)
-[![Scaladoc](http://javadoc-badge.appspot.com/com.github.j5ik2o/akka-persistence-kafka_2.12.svg?label=scaladoc)](http://javadoc-badge.appspot.com/com.github.j5ik2o/akka-persistence-kafka_2.12/com/github/j5ik2o/akka/persistence/kafka/index.html?javadocio=true)
+[![CircleCI](https://circleci.com/gh/pawelkaczor/akka-persistence-kafka/tree/master.svg?style=shield&circle-token=c809688daf71f6ae582dd2d58cb5518401498373)](https://circleci.com/gh/pawelkaczor/akka-persistence-kafka/tree/master)
+[![Maven Central](https://maven-badges.herokuapp.com/maven-central/pl.newicom/akka-persistence-kafka_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/pl.newicom/akka-persistence-kafka_2.12)
+[![Scaladoc](http://javadoc-badge.appspot.com/pl.newicom/akka-persistence-kafka_2.12.svg?label=scaladoc)](http://javadoc-badge.appspot.com/pl.newicom/akka-persistence-kafka_2.12/pl/newicom/akka/persistence/kafka/index.html?javadocio=true)
 [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
 
 akka-persistence-kafka writes journal and snapshot entries to Kafka.
@@ -27,7 +27,7 @@ resolvers += "Sonatype OSS Release Repository" at "https://oss.sonatype.org/cont
 val version = "..."
 
 libraryDependencies += Seq(
-  "com.github.j5ik2o" %% "akka-persistence-kafka" % version
+  "pl.newicom" %% "akka-persistence-kafka" % version
 )
 ```
 
@@ -45,9 +45,9 @@ j5ik2o {
   kafka-journal {
     topic-prefix = "journal-"
     # if need customize, default is persistence-id
-    topic-resolver-class-name = "com.github.j5ik2o.akka.persistence.kafka.resolver.KafkaTopicResolver$PersistenceId"
+    topic-resolver-class-name = "pl.newicom.akka.persistence.kafka.resolver.KafkaTopicResolver$PersistenceId"
     # if need customize, default is partion 0
-    partition-resolver-class-name = "com.github.j5ik2o.akka.persistence.kafka.resolver.KafkaPartitionResolver$PartitionZero"
+    partition-resolver-class-name = "pl.newicom.akka.persistence.kafka.resolver.KafkaPartitionResolver$PartitionZero"
 
     producer {
       kafka-clients {
@@ -65,9 +65,9 @@ j5ik2o {
   kafka-snapshot-store {
     topic-prefix = "snapshot-"
     # if need customize, default is persistence-id
-    topic-resolver-class-name = "com.github.j5ik2o.akka.persistence.kafka.resolver.KafkaTopicResolver$PersistenceId"
+    topic-resolver-class-name = "pl.newicom.akka.persistence.kafka.resolver.KafkaTopicResolver$PersistenceId"
     # if need customize, default is partition 0
-    partition-resolver-class-name = "com.github.j5ik2o.akka.persistence.kafka.resolver.KafkaPartitionResolver$PartitionZero"
+    partition-resolver-class-name = "pl.newicom.akka.persistence.kafka.resolver.KafkaPartitionResolver$PartitionZero"
 
     producer {
       kafka-clients {

build.sbt

@@ -6,8 +6,8 @@ val kafkaVersion        = "2.4.1.1"
 val alpakkaKafkaVersion = "2.0.2"
 
 val coreSettings = Seq(
-  sonatypeProfileName := "com.github.j5ik2o",
-  organization := "com.github.j5ik2o",
+  sonatypeProfileName := "pl.newicom",
+  organization := "pl.newicom",
   scalaVersion := scala213Version,
   crossScalaVersions ++= Seq(scala212Version, scala213Version),
   scalacOptions ++= {
@@ -42,6 +42,10 @@ val coreSettings = Seq(
         <id>j5ik2o</id>
         <name>Junichi Kato</name>
       </developer>
+      <developer>
+        <id>pawelkaczor</id>
+        <name>Pawel Kaczor</name>
+      </developer>
     </developers>
   },
   publishTo := sonatypePublishToBundle.value,
@@ -51,45 +55,45 @@ val coreSettings = Seq(
     Credentials(ivyCredentials) :: Credentials(gpgCredentials) :: Nil
   },
   resolvers ++= Seq(
-      "Sonatype OSS Snapshot Repository" at "https://oss.sonatype.org/content/repositories/snapshots/",
-      "Sonatype OSS Release Repository" at "https://oss.sonatype.org/content/repositories/releases/"
-    ),
+    "Sonatype OSS Snapshot Repository" at "https://oss.sonatype.org/content/repositories/snapshots/",
+    "Sonatype OSS Release Repository" at "https://oss.sonatype.org/content/repositories/releases/"
+  ),
   libraryDependencies ++= Seq(
-      "org.scala-lang"          % "scala-reflect"      % scalaVersion.value,
-      "com.iheart"              %% "ficus"             % "1.4.7",
-      "org.slf4j"               % "slf4j-api"          % "1.7.30",
-      "com.typesafe.akka"       %% "akka-stream-kafka" % alpakkaKafkaVersion,
-      "com.thesamet.scalapb"    %% "scalapb-runtime"   % scalapb.compiler.Version.scalapbVersion % "protobuf",
-      "ch.qos.logback"          % "logback-classic"    % "1.2.3" % Test,
-      "io.github.embeddedkafka" %% "embedded-kafka"    % kafkaVersion % Test
-    ) ++ {
-      CrossVersion.partialVersion(scalaVersion.value) match {
-        case Some((2L, scalaMajor)) if scalaMajor == 13 =>
-          Seq(
-            "com.typesafe.akka" %% "akka-slf4j"           % akka26Version,
-            "com.typesafe.akka" %% "akka-stream"          % akka26Version,
-            "com.typesafe.akka" %% "akka-persistence"     % akka26Version,
-            "com.typesafe.akka" %% "akka-testkit"         % akka26Version % Test,
-            "com.typesafe.akka" %% "akka-stream-testkit"  % akka26Version % Test,
-            "com.typesafe.akka" %% "akka-persistence-tck" % akka26Version % Test,
-            "org.scalatest"     %% "scalatest"            % "3.1.1" % Test
-          )
-        case Some((2L, scalaMajor)) if scalaMajor == 12 =>
-          Seq(
-            "org.scala-lang.modules" %% "scala-collection-compat" % "2.1.6",
-            "com.typesafe.akka"      %% "akka-slf4j"              % akka26Version,
-            "com.typesafe.akka"      %% "akka-stream"             % akka26Version,
-            "com.typesafe.akka"      %% "akka-persistence"        % akka26Version,
-            "com.typesafe.akka"      %% "akka-testkit"            % akka26Version % Test,
-            "com.typesafe.akka"      %% "akka-stream-testkit"     % akka26Version % Test,
-            "com.typesafe.akka"      %% "akka-persistence-tck"    % akka26Version % Test,
-            "org.scalatest"          %% "scalatest"               % "3.1.1" % Test
-          )
-      }
-    },
+    "org.scala-lang"           % "scala-reflect"     % scalaVersion.value,
+    "com.iheart"              %% "ficus"             % "1.4.7",
+    "org.slf4j"                % "slf4j-api"         % "1.7.30",
+    "com.typesafe.akka"       %% "akka-stream-kafka" % alpakkaKafkaVersion,
+    "com.thesamet.scalapb"    %% "scalapb-runtime"   % scalapb.compiler.Version.scalapbVersion % "protobuf",
+    "ch.qos.logback"           % "logback-classic"   % "1.2.3"                                 % Test,
+    "io.github.embeddedkafka" %% "embedded-kafka"    % kafkaVersion                            % Test
+  ) ++ {
+    CrossVersion.partialVersion(scalaVersion.value) match {
+      case Some((2L, scalaMajor)) if scalaMajor == 13 =>
+        Seq(
+          "com.typesafe.akka" %% "akka-slf4j"           % akka26Version,
+          "com.typesafe.akka" %% "akka-stream"          % akka26Version,
+          "com.typesafe.akka" %% "akka-persistence"     % akka26Version,
+          "com.typesafe.akka" %% "akka-testkit"         % akka26Version % Test,
+          "com.typesafe.akka" %% "akka-stream-testkit"  % akka26Version % Test,
+          "com.typesafe.akka" %% "akka-persistence-tck" % akka26Version % Test,
+          "org.scalatest"          %% "scalatest"               % "3.1.4" % Test
+        )
+      case Some((2L, scalaMajor)) if scalaMajor == 12 =>
+        Seq(
+          "org.scala-lang.modules" %% "scala-collection-compat" % "2.1.6",
+          "com.typesafe.akka"      %% "akka-slf4j"              % akka26Version,
+          "com.typesafe.akka"      %% "akka-stream"             % akka26Version,
+          "com.typesafe.akka"      %% "akka-persistence"        % akka26Version,
+          "com.typesafe.akka"      %% "akka-testkit"            % akka26Version % Test,
+          "com.typesafe.akka"      %% "akka-stream-testkit"     % akka26Version % Test,
+          "com.typesafe.akka"      %% "akka-persistence-tck"    % akka26Version % Test,
+          "org.scalatest"          %% "scalatest"               % "3.1.4" % Test
+        )
+    }
+  },
   PB.targets in Compile := Seq(
-      scalapb.gen() -> (sourceManaged in Compile).value
-    ),
+    scalapb.gen() -> (sourceManaged in Compile).value
+  ),
   parallelExecution in Test := false
 )
 

project/build.properties

@@ -1 +1 @@
-sbt.version = 1.3.12
+sbt.version = 1.4.0

project/plugins.sbt

@@ -15,6 +15,6 @@ addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.31")
 
 addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.13")
 
-addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "3.9.2")
+addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "3.9.4")
 
 addSbtPlugin("com.jsuereth" % "sbt-pgp" % "2.0.1")

src/main/scala/com/github/j5ik2o/akka/persistence/kafka/journal/JournalSequence.scala

@@ -14,13 +14,13 @@ class JournalSequence(
     journalPartitionResolver: KafkaPartitionResolver
 ) {
 
-  def readLowestSequenceNrAsync(persistenceId: PersistenceId, fromSequenceNr: Option[Long] = None)(
-      implicit ec: ExecutionContext
+  def readLowestSequenceNrAsync(persistenceId: PersistenceId, fromSequenceNr: Option[Long] = None)(implicit
+      ec: ExecutionContext
   ): Future[Long] =
     Future { readLowestSequenceNr(persistenceId, fromSequenceNr) }
 
-  def readHighestSequenceNrAsync(persistenceId: PersistenceId, fromSequenceNr: Option[Long] = None)(
-      implicit ec: ExecutionContext
+  def readHighestSequenceNrAsync(persistenceId: PersistenceId, fromSequenceNr: Option[Long] = None)(implicit
+      ec: ExecutionContext
   ): Future[Long] =
     Future { readHighestSequenceNr(persistenceId, fromSequenceNr) }
 

src/main/scala/com/github/j5ik2o/akka/persistence/kafka/journal/KafkaJournal.scala

@@ -162,15 +162,14 @@ class KafkaJournal(config: Config) extends AsyncWriteJournal with ActorLogging {
             )
           )
         } else
-          ProducerMessage.multi(rowsToWrite.map {
-            case (journal, byteArray) =>
-              new ProducerRecord(
-                resolveTopic(journal.persistenceId),
-                resolvePartition(journal.persistenceId),
-                journal.persistenceId.asString,
-                byteArray,
-                createHeaders(journal)
-              )
+          ProducerMessage.multi(rowsToWrite.map { case (journal, byteArray) =>
+            new ProducerRecord(
+              resolveTopic(journal.persistenceId),
+              resolvePartition(journal.persistenceId),
+              journal.persistenceId.asString,
+              byteArray,
+              createHeaders(journal)
+            )
           }.asJava) // asJava method, Must not modify for 2.12
       val future: Future[immutable.Seq[Try[Unit]]] = Source
         .single(messages)
@@ -207,10 +206,11 @@ class KafkaJournal(config: Config) extends AsyncWriteJournal with ActorLogging {
   override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long): Future[Unit] = {
     log.debug(s"asyncDeleteMessagesTo($persistenceId, $toSequenceNr): start")
     val future = for {
-      to <- if (toSequenceNr == Long.MaxValue)
-        journalSequence.readHighestSequenceNrAsync(PersistenceId(persistenceId))
-      else
-        Future.successful(toSequenceNr)
+      to <-
+        if (toSequenceNr == Long.MaxValue)
+          journalSequence.readHighestSequenceNrAsync(PersistenceId(persistenceId))
+        else
+          Future.successful(toSequenceNr)
       _ <- Future {
         adminClient
           .deleteRecords(
@@ -279,23 +279,21 @@ class KafkaJournal(config: Config) extends AsyncWriteJournal with ActorLogging {
               .fromBinaryAsync(record.value(), classOf[JournalRow].getName)
               .map(journal => (record, journal.asInstanceOf[JournalRow]))
           }
-          .map {
-            case (record, journal) =>
-              log.debug(s"record = $record, journal = $journal")
-              (
-                record,
-                journal.persistentRepr
-              )
+          .map { case (record, journal) =>
+            log.debug(s"record = $record, journal = $journal")
+            (
+              record,
+              journal.persistentRepr
+            )
           }
-          .map {
-            case (record, persistentRepr) =>
-              log.debug(
-                s"record.offset = ${record.offset()}, persistentRepr.sequenceNr = ${persistentRepr.sequenceNr}, deletedTo = $deletedTo"
-              )
-              if (persistentRepr.sequenceNr <= deletedTo) {
-                log.debug("update: deleted = true")
-                persistentRepr.update(deleted = true)
-              } else persistentRepr
+          .map { case (record, persistentRepr) =>
+            log.debug(
+              s"record.offset = ${record.offset()}, persistentRepr.sequenceNr = ${persistentRepr.sequenceNr}, deletedTo = $deletedTo"
+            )
+            if (persistentRepr.sequenceNr <= deletedTo) {
+              log.debug("update: deleted = true")
+              persistentRepr.update(deleted = true)
+            } else persistentRepr
           }
           .runWith(Sink.foreach { persistentRepr =>
             if (adjustedFrom <= persistentRepr.sequenceNr && persistentRepr.sequenceNr <= adjustedTo) {

src/main/scala/com/github/j5ik2o/akka/persistence/kafka/serialization/EncryptionFilter.scala

@@ -21,12 +21,12 @@ trait EncryptionFilter {
 object EncryptionFilter {
 
   class NonEncryption extends EncryptionFilter {
-    override def encrypt(persistenceId: PersistenceId, bytes: Array[Byte], context: Map[String, AnyRef])(
-        implicit ec: ExecutionContext
+    override def encrypt(persistenceId: PersistenceId, bytes: Array[Byte], context: Map[String, AnyRef])(implicit
+        ec: ExecutionContext
     ): Future[Array[Byte]] = Future.successful(bytes)
 
-    override def decrypt(bytes: Array[Byte])(
-        implicit ec: ExecutionContext
+    override def decrypt(bytes: Array[Byte])(implicit
+        ec: ExecutionContext
     ): Future[Array[Byte]] = Future.successful(bytes)
   }
 

src/main/scala/com/github/j5ik2o/akka/persistence/kafka/serialization/PersistentReprSerializer.scala

@@ -39,8 +39,8 @@ class PersistentReprSerializer(serialization: Serialization) {
     serializer.toBinaryAsync(journal).map((journal, _))
   }
 
-  def serialize(persistentRepr: PersistentRepr, index: Option[Int])(
-      implicit ec: ExecutionContext
+  def serialize(persistentRepr: PersistentRepr, index: Option[Int])(implicit
+      ec: ExecutionContext
   ): Future[JournalWithByteArray] = {
     persistentRepr.payload match {
       case Tagged(payload, tags) =>
@@ -54,8 +54,8 @@ class PersistentReprSerializer(serialization: Serialization) {
       atomicWrites: Seq[AtomicWrite]
   )(implicit ec: ExecutionContext): Seq[Future[Seq[JournalWithByteArray]]] = {
     atomicWrites.map { atomicWrite =>
-      val serialized = atomicWrite.payload.zipWithIndex.map {
-        case (v, index) => serialize(v, Some(index))
+      val serialized = atomicWrite.payload.zipWithIndex.map { case (v, index) =>
+        serialize(v, Some(index))
       }
       Future.sequence(serialized)
     }

src/main/scala/com/github/j5ik2o/akka/persistence/kafka/snapshot/KafkaSnapshotStore.scala

@@ -135,14 +135,18 @@ class KafkaSnapshotStore(config: Config) extends SnapshotStore {
     val rangeDeletions  = this.rangeDeletions
 
     for {
-      highest <- if (ignoreOrphan)
-        journalSequence.readHighestSequenceNrAsync(PersistenceId(persistenceId))
-      else
-        Future.successful(Long.MaxValue)
-      adjusted = if (ignoreOrphan &&
-                     highest < criteria.maxSequenceNr &&
-                     highest > 0L) criteria.copy(maxSequenceNr = highest)
-      else criteria
+      highest <-
+        if (ignoreOrphan)
+          journalSequence.readHighestSequenceNrAsync(PersistenceId(persistenceId))
+        else
+          Future.successful(Long.MaxValue)
+      adjusted =
+        if (
+          ignoreOrphan &&
+          highest < criteria.maxSequenceNr &&
+          highest > 0L
+        ) criteria.copy(maxSequenceNr = highest)
+        else criteria
       // highest  <- Future.successful(Long.MaxValue)
       // adjusted = criteria
       snapshot <- {

src/test/scala/com/github/j5ik2o/akka/persistence/kafka/journal/KafkaJournalPerfSpec.scala

@@ -8,9 +8,11 @@ import org.scalatest.BeforeAndAfterAll
 
 class KafkaJournalPerfSpec
     extends JournalPerfSpec(
-      ConfigFactory.parseString("""
+      ConfigFactory
+        .parseString("""
                               |akka.test.single-expect-default = 60s
-      """.stripMargin).withFallback(ConfigFactory.load())
+      """.stripMargin)
+        .withFallback(ConfigFactory.load())
     )
     with BeforeAndAfterAll {