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/AlpakkaCommittableProducer.scala → benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/AlpakkaCommittableProducer.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
 
 /**
  * Compares the `CommittingProducerSinkStage` with the composed implementation of `Producer.flexiFlow` and `Committer.sink`.

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 {

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 {

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._
 

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"

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() {

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

@@ -3,18 +3,18 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
-
-import akka.Done
-import akka.actor.ActorSystem
-import akka.kafka.ConsumerMessage.{ Committable, CommittableMessage }
-import akka.kafka.ProducerMessage.Envelope
-import akka.kafka.benchmarks.app.RunTestCommand
-import akka.kafka.scaladsl.Consumer.{ Control, DrainingControl }
-import akka.kafka.scaladsl.{ Committer, Consumer, Producer }
-import akka.kafka._
-import akka.stream.Materializer
-import akka.stream.scaladsl.{ Keep, Sink, Source }
+package org.apache.pekko.kafka.benchmarks
+
+import org.apache.pekko.Done
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableMessage }
+import org.apache.pekko.kafka.ProducerMessage.Envelope
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
+import org.apache.pekko.kafka.scaladsl.{ Committer, Consumer, Producer }
+import org.apache.pekko.kafka._
+import org.apache.pekko.stream.Materializer
+import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
 import com.codahale.metrics.Meter
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.kafka.clients.consumer.ConsumerConfig

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

@@ -0,0 +1,116 @@
+/*
+ * 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 akka.stream.alpakka.csv.scaladsl.CsvQuotingStyle
+import org.apache.pekko.util.ByteString
+
+import java.nio.charset.{ Charset, StandardCharsets }
+import scala.collection.immutable
+
+// 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)

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

@@ -3,15 +3,15 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
 import java.lang.management.{ BufferPoolMXBean, ManagementFactory, MemoryType }
 
-import akka.NotUsed
-import akka.actor.Cancellable
-import akka.kafka.scaladsl.Consumer.Control
-import akka.stream.Materializer
-import akka.stream.scaladsl.{ Keep, Sink, Source }
+import org.apache.pekko.NotUsed
+import org.apache.pekko.actor.Cancellable
+import org.apache.pekko.kafka.scaladsl.Consumer.Control
+import org.apache.pekko.stream.Materializer
+import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
 import com.codahale.metrics.{ Histogram, MetricRegistry }
 import javax.management.remote.{ JMXConnectorFactory, JMXServiceURL }
 import javax.management.{ Attribute, MBeanServerConnection, ObjectName }

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

@@ -3,7 +3,7 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
 import java.time.Duration
 import java.util

benchmarks/src/main/scala/akka/kafka/benchmarks/KafkaConsumerFixtureGen.scala → benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerFixtureGen.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.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 import org.apache.kafka.clients.consumer.{ ConsumerConfig, KafkaConsumer }
 import org.apache.kafka.common.serialization.{ ByteArrayDeserializer, StringDeserializer }
 

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

@@ -3,7 +3,7 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
 import com.codahale.metrics.Meter
 import com.typesafe.scalalogging.LazyLogging

benchmarks/src/main/scala/akka/kafka/benchmarks/KafkaProducerFixtureGen.scala → benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerFixtureGen.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.PerfFixtureHelpers.FilledTopic
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 import org.apache.kafka.clients.producer.KafkaProducer
 
 case class KafkaProducerTestFixture(topic: String,

benchmarks/src/main/scala/akka/kafka/benchmarks/KafkaTransactionBenchmarks.scala → benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.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.benchmarks.KafkaConsumerBenchmarks.pollTimeoutMs
+import org.apache.pekko.kafka.benchmarks.KafkaConsumerBenchmarks.pollTimeoutMs
 import com.codahale.metrics.Meter
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.kafka.clients.consumer._

benchmarks/src/main/scala/akka/kafka/benchmarks/KafkaTransactionFixtureGen.scala → benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.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 java.util.Locale
 
-import akka.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
 import org.apache.kafka.clients.consumer.{ ConsumerConfig, KafkaConsumer }
 import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerConfig }
 import org.apache.kafka.common.IsolationLevel

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

@@ -3,7 +3,7 @@
  * Copyright (C) 2016 - 2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.kafka.benchmarks
+package org.apache.pekko.kafka.benchmarks
 
 import java.time.Duration
 import java.util