Support Alpakka Kafka snapshot and troubleshoot Kafka integration tes…

…ts (#200)

.travis.yml

@@ -16,7 +16,7 @@ script:
   - jabba install ${JDK:=adopt@~1.8-0}
   - jabba use ${JDK:=adopt@~1.8-0}
   - java -version
-  - sbt -jvm-opts .jvmopts-travis "$AKKA_SNAPSHOT" "$CMD"
+  - sbt -jvm-opts .jvmopts-travis "$AKKA_SNAPSHOT" "$ALPAKKA_KAFKA_SNAPSHOT" "$CMD"
 
 jobs:
   include:
@@ -30,9 +30,11 @@ jobs:
 
     - stage: test
       env:
+        - ALPAKKA_KAFKA_SNAPSHOT="-Dbuild.alpakka.kafka.version=2.0.3+7-edfcbb02" # Remove when Alpakka Kafka 2.0.4 released
         - CMD="test"
       name: "Run tests with Scala 2.13 and AdoptOpenJDK 8"
     - env:
+        - ALPAKKA_KAFKA_SNAPSHOT="-Dbuild.alpakka.kafka.version=2.0.3+7-edfcbb02" # Remove when Alpakka Kafka 2.0.4 released
         - JDK="adopt@~1.11-0"
         - _JAVA_OPTIONS="-XX:+UnlockExperimentalVMOptions -XX:+UseJVMCICompiler"
         - CMD="test"

akka-projection-kafka/src/test/resources/application.conf

@@ -2,7 +2,7 @@ akka {
   loggers = ["akka.event.slf4j.Slf4jLogger"]
   loglevel = "DEBUG"
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
-  logger-startup-timeout = 15s
+  logger-startup-timeout = 15 seconds
 
   actor {
     debug {
@@ -11,11 +11,7 @@ akka {
     }
   }
 
-  test {
-    single-expect-default = 5 seconds
-  }
-
   kafka.consumer {
-    stop-timeout = 3 s
+    stop-timeout = 3 seconds
   }
 }

akka-projection-kafka/src/test/scala/akka/projection/kafka/integration/KafkaToSlickIntegrationSpec.scala

@@ -135,10 +135,9 @@ class KafkaToSlickIntegrationSpec extends KafkaSpecBase(ConfigFactory.load().wit
 
   "KafkaSourceProvider with Slick" must {
     "project a model and Kafka offset map to a slick db exactly once" in {
-      val projectionId = ProjectionId("HappyPath", "UserEventCountProjection-1")
-
       val topicName = createTopic(suffix = 0, partitions = 3, replication = 1)
       val groupId = createGroupId()
+      val projectionId = ProjectionId(groupId, "UserEventCountProjection-1")
 
       produceEvents(topicName)
 
@@ -166,10 +165,9 @@ class KafkaToSlickIntegrationSpec extends KafkaSpecBase(ConfigFactory.load().wit
     }
 
     "project a model and Kafka offset map to a slick db exactly once with a retriable DBIO.failed" in {
-      val projectionId = ProjectionId("OneFailure", "UserEventCountProjection-1")
-
       val topicName = createTopic(suffix = 1, partitions = 3, replication = 1)
       val groupId = createGroupId()
+      val projectionId = ProjectionId(groupId, "UserEventCountProjection-1")
 
       produceEvents(topicName)
 

project/Dependencies.scala

@@ -16,7 +16,7 @@ object Dependencies {
   object Versions {
     val akka = sys.props.getOrElse("build.akka.version", "2.6.6")
     val alpakka = "2.0.0"
-    val alpakkaKafka = "2.0.3"
+    val alpakkaKafka = sys.props.getOrElse("build.alpakka.kafka.version", "2.0.3")
     val slick = "3.3.2"
     val scalaTest = "3.1.1"
     val testContainersScala = "0.37.0"