Pekko migration

.scalafmt.conf

@@ -35,6 +35,7 @@ optIn.configStyleArguments               = false
 danglingParentheses.preset               = false
 spaces.inImportCurlyBraces               = true
 rewrite.neverInfix.excludeFilters        = [
+  at
   and
   min
   max

benchmarks/src/it/scala/akka/kafka/benchmarks/BatchedConsumer.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/BatchedConsumer.scala

@@ -3,11 +3,11 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.BenchmarksBase.{ topic_1000_100, topic_1000_5000, topic_1000_5000_8 }
-import akka.kafka.benchmarks.Timed.runPerfTest
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.BenchmarksBase.{ topic_1000_100, topic_1000_5000, topic_1000_5000_8 }
+import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 
 class ApacheKafkaBatchedConsumer extends BenchmarksBase() {
   it should "bench with small messages" in {
@@ -36,7 +36,7 @@ class ApacheKafkaBatchedConsumer extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaBatchedConsumer extends BenchmarksBase() {
+class PekkoConnectorsKafkaBatchedConsumer extends BenchmarksBase() {
 
   it should "bench with small messages" in {
     val cmd = RunTestCommand("alpakka-kafka-batched-consumer", bootstrapServers, topic_1000_100)

benchmarks/src/it/scala/akka/kafka/benchmarks/Benchmarks.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Benchmarks.scala

@@ -3,15 +3,15 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
-
-import akka.kafka.benchmarks.BenchmarksBase._
-import akka.kafka.benchmarks.InflightMetrics._
-import akka.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
-import akka.kafka.benchmarks.Timed.{ runPerfTest, runPerfTestInflightMetrics }
-import akka.kafka.benchmarks.app.RunTestCommand
-import akka.kafka.testkit.KafkaTestkitTestcontainersSettings
-import akka.kafka.testkit.scaladsl.TestcontainersKafkaLike
+package org.apache.pekko.kafka.benchmarks
+
+import org.apache.pekko.kafka.benchmarks.BenchmarksBase._
+import org.apache.pekko.kafka.benchmarks.InflightMetrics._
+import org.apache.pekko.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
+import org.apache.pekko.kafka.benchmarks.Timed.{ runPerfTest, runPerfTestInflightMetrics }
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
 import com.typesafe.config.Config
 
 object BenchmarksBase {
@@ -61,7 +61,7 @@ class ApacheKafkaConsumerNokafka extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaConsumerNokafka extends BenchmarksBase() {
+class PekkoConnectorsKafkaConsumerNokafka extends BenchmarksBase() {
   it should "bench" in {
     val cmd = RunTestCommand("alpakka-kafka-plain-consumer-nokafka", bootstrapServers, topic_2000_100)
     runPerfTest(cmd,
@@ -83,7 +83,7 @@ class ApacheKafkaPlainConsumer extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaPlainConsumer extends BenchmarksBase() {
+class PekkoConnectorsKafkaPlainConsumer extends BenchmarksBase() {
   it should "bench" in {
     val cmd = RunTestCommand("alpakka-kafka-plain-consumer", bootstrapServers, topic_2000_100)
     runPerfTest(cmd, ReactiveKafkaConsumerFixtures.plainSources(cmd), ReactiveKafkaConsumerBenchmarks.consumePlain)
@@ -127,7 +127,7 @@ class ApacheKafkaAtMostOnceConsumer extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaAtMostOnceConsumer extends BenchmarksBase() {
+class PekkoConnectorsKafkaAtMostOnceConsumer extends BenchmarksBase() {
   it should "bench" in {
     val cmd = RunTestCommand("alpakka-kafka-at-most-once-consumer", bootstrapServers, topic_50_100)
     runPerfTest(cmd,

benchmarks/src/it/scala/akka/kafka/benchmarks/NoCommitBackpressure.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/NoCommitBackpressure.scala

@@ -3,10 +3,10 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.Timed.runPerfTest
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 
 import BenchmarksBase._
 
@@ -38,7 +38,7 @@ class RawKafkaCommitEveryPollConsumer extends BenchmarksBase() {
 //  }
 }
 
-class AlpakkaCommitAndForgetConsumer extends BenchmarksBase() {
+class PekkoConnectorsCommitAndForgetConsumer extends BenchmarksBase() {
   val prefix = "alpakka-kafka-commit-and-forget-"
 
   it should "bench with small messages" in {

benchmarks/src/it/scala/akka/kafka/benchmarks/AlpakkaCommittableProducer.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableProducer.scala

@@ -3,45 +3,45 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
-import akka.kafka.benchmarks.Timed.runPerfTest
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
+import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 
 /**
  * Compares the `CommittingProducerSinkStage` with the composed implementation of `Producer.flexiFlow` and `Committer.sink`.
  */
-class AlpakkaCommittableProducer extends BenchmarksBase() {
+class PekkoConnectorsCommittableProducer extends BenchmarksBase() {
   it should "bench composed sink with 100b messages" in {
     val cmd = RunTestCommand("alpakka-committable-producer-composed", bootstrapServers, topic_100_100)
     runPerfTest(
       cmd,
-      AlpakkaCommittableSinkFixtures.composedSink(cmd),
-      AlpakkaCommittableSinkBenchmarks.run)
+      PekkoConnectorsCommittableSinkFixtures.composedSink(cmd),
+      PekkoConnectorsCommittableSinkBenchmarks.run)
   }
 
   it should "bench composed sink with 5000b messages" in {
     val cmd = RunTestCommand("alpakka-committable-producer-composed-5000b", bootstrapServers, topic_100_5000)
     runPerfTest(
       cmd,
-      AlpakkaCommittableSinkFixtures.composedSink(cmd),
-      AlpakkaCommittableSinkBenchmarks.run)
+      PekkoConnectorsCommittableSinkFixtures.composedSink(cmd),
+      PekkoConnectorsCommittableSinkBenchmarks.run)
   }
 
   it should "bench `Producer.committableSink` with 100b messages" in {
     val cmd = RunTestCommand("alpakka-committable-producer", bootstrapServers, topic_100_100)
     runPerfTest(
       cmd,
-      AlpakkaCommittableSinkFixtures.producerSink(cmd),
-      AlpakkaCommittableSinkBenchmarks.run)
+      PekkoConnectorsCommittableSinkFixtures.producerSink(cmd),
+      PekkoConnectorsCommittableSinkBenchmarks.run)
   }
 
   it should "bench `Producer.committableSink` with 5000b messages" in {
     val cmd = RunTestCommand("alpakka-committable-producer-5000b", bootstrapServers, topic_100_5000)
     runPerfTest(
       cmd,
-      AlpakkaCommittableSinkFixtures.producerSink(cmd),
-      AlpakkaCommittableSinkBenchmarks.run)
+      PekkoConnectorsCommittableSinkFixtures.producerSink(cmd),
+      PekkoConnectorsCommittableSinkBenchmarks.run)
   }
 }

benchmarks/src/it/scala/akka/kafka/benchmarks/Producer.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Producer.scala

@@ -3,11 +3,16 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.BenchmarksBase.{ topic_2000_100, topic_2000_500, topic_2000_5000, topic_2000_5000_8 }
-import akka.kafka.benchmarks.Timed.runPerfTest
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.BenchmarksBase.{
+  topic_2000_100,
+  topic_2000_500,
+  topic_2000_5000,
+  topic_2000_5000_8
+}
+import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 
 class ApacheKafkaPlainProducer extends BenchmarksBase() {
   private val prefix = "apache-kafka-plain-producer"
@@ -34,7 +39,7 @@ class ApacheKafkaPlainProducer extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaPlainProducer extends BenchmarksBase() {
+class PekkoConnectorsKafkaPlainProducer extends BenchmarksBase() {
   private val prefix = "alpakka-kafka-plain-producer"
 
   it should "bench with small messages" in {

benchmarks/src/it/scala/akka/kafka/benchmarks/SpecBase.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/SpecBase.scala

@@ -3,9 +3,9 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.testkit.scaladsl.ScalatestKafkaSpec
+import org.apache.pekko.kafka.testkit.scaladsl.ScalatestKafkaSpec
 import org.scalatest.concurrent.{ Eventually, ScalaFutures }
 import org.scalatest.flatspec.AnyFlatSpecLike
 import org.scalatest.matchers.should.Matchers

benchmarks/src/it/scala/akka/kafka/benchmarks/Transactions.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Transactions.scala

@@ -3,11 +3,11 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
-import akka.kafka.benchmarks.Timed.runPerfTest
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
+import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 import scala.concurrent.duration._
 
 class ApacheKafkaTransactions extends BenchmarksBase() {
@@ -26,7 +26,7 @@ class ApacheKafkaTransactions extends BenchmarksBase() {
   }
 }
 
-class AlpakkaKafkaTransactions extends BenchmarksBase() {
+class PekkoConnectorsKafkaTransactions extends BenchmarksBase() {
   it should "bench with small messages" in {
     val cmd = RunTestCommand("alpakka-kafka-transactions", bootstrapServers, topic_100_100)
     runPerfTest(

benchmarks/src/main/resources/logback.xml

@@ -13,8 +13,8 @@
     </appender>
 
     <logger name="org.apache" level="WARN"/>
-    <logger name="akka" level="WARN"/>
-    <logger name="akka.kafka.benchmarks" level="INFO"/>
+    <logger name="org.apache.pekko" level="WARN"/>
+    <logger name="org.apache.pekko.kafka.benchmarks" level="INFO"/>
     <logger name="org.apache.kafka.common.utils.AppInfoParser" level="ERROR"/>
     <logger name="org.apache.kafka.clients.NetworkClient" level="ERROR"/>
 

benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/CsvFormatter.scala

@@ -0,0 +1,126 @@
+/*
+ * Copyright (C) 2014 - 2016 Softwaremill <https://softwaremill.com>
+ * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.kafka.benchmarks
+
+import org.apache.pekko.util.ByteString
+
+import java.nio.charset.{ Charset, StandardCharsets }
+import scala.collection.immutable
+
+private[benchmarks] sealed trait CsvQuotingStyle
+
+object CsvQuotingStyle {
+
+  /** Quote only fields requiring quotes */
+  case object Required extends CsvQuotingStyle
+
+  /** Quote all fields */
+  case object Always extends CsvQuotingStyle
+}
+
+// TODO: This needs to be deleted after migrating alpakka to pekko.
+// This is just temporary base to see everything compiles and tests will pass without any issue
+private[benchmarks] class CsvFormatter(delimiter: Char,
+    quoteChar: Char,
+    escapeChar: Char,
+    endOfLine: String,
+    quotingStyle: CsvQuotingStyle,
+    charset: Charset = StandardCharsets.UTF_8) {
+
+  private[this] val charsetName = charset.name()
+
+  private[this] val delimiterBs = ByteString(String.valueOf(delimiter), charsetName)
+  private[this] val quoteBs = ByteString(String.valueOf(quoteChar), charsetName)
+  private[this] val duplicatedQuote = ByteString(String.valueOf(Array(quoteChar, quoteChar)), charsetName)
+  private[this] val duplicatedEscape = ByteString(String.valueOf(Array(escapeChar, escapeChar)), charsetName)
+  private[this] val endOfLineBs = ByteString(endOfLine, charsetName)
+
+  def toCsv(fields: immutable.Iterable[Any]): ByteString =
+    if (fields.nonEmpty) nonEmptyToCsv(fields)
+    else endOfLineBs
+
+  private def nonEmptyToCsv(fields: immutable.Iterable[Any]) = {
+    val builder = ByteString.createBuilder
+
+    def splitAndDuplicateQuotesAndEscapes(field: String, splitAt: Int) = {
+
+      @inline def indexOfQuoteOrEscape(lastIndex: Int) = {
+        var index = lastIndex
+        var found = -1
+        while (index < field.length && found == -1) {
+          val char = field(index)
+          if (char == quoteChar || char == escapeChar) found = index
+          index += 1
+        }
+        found
+      }
+
+      var lastIndex = 0
+      var index = splitAt
+      while (index > -1) {
+        builder ++= ByteString.apply(field.substring(lastIndex, index), charsetName)
+        val char = field.charAt(index)
+        if (char == quoteChar) {
+          builder ++= duplicatedQuote
+        } else {
+          builder ++= duplicatedEscape
+        }
+        lastIndex = index + 1
+        index = indexOfQuoteOrEscape(lastIndex)
+      }
+      if (lastIndex < field.length) {
+        builder ++= ByteString(field.substring(lastIndex), charsetName)
+      }
+    }
+
+    def append(field: String) = {
+      val (quoteIt, splitAt) = requiresQuotesOrSplit(field)
+      if (quoteIt || quotingStyle == CsvQuotingStyle.Always) {
+        builder ++= quoteBs
+        if (splitAt != -1) {
+          splitAndDuplicateQuotesAndEscapes(field, splitAt)
+        } else {
+          builder ++= ByteString(field, charsetName)
+        }
+        builder ++= quoteBs
+      } else {
+        builder ++= ByteString(field, charsetName)
+      }
+    }
+
+    val iterator = fields.iterator
+    var hasNext = iterator.hasNext
+    while (hasNext) {
+      val next = iterator.next()
+      if (next != null) {
+        append(next.toString)
+      }
+      hasNext = iterator.hasNext
+      if (hasNext) {
+        builder ++= delimiterBs
+      }
+    }
+    builder ++= endOfLineBs
+    builder.result()
+  }
+
+  private def requiresQuotesOrSplit(field: String): (Boolean, Int) = {
+    var quotes = CsvQuotingStyle.Always == quotingStyle
+    var split = -1
+    var index = 0
+    while (index < field.length && !(quotes && split != -1)) {
+      val char = field(index)
+      if (char == `quoteChar` || char == `escapeChar`) {
+        quotes = true
+        split = index
+      } else if (char == '\r' || char == '\n' || char == `delimiter`) {
+        quotes = true
+      }
+      index += 1
+    }
+    (quotes, split)
+  }
+}

benchmarks/src/main/scala/akka/kafka/benchmarks/FixtureGen.scala → benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/FixtureGen.scala

@@ -3,8 +3,8 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 
 case class FixtureGen[F](command: RunTestCommand, generate: Int => F)