You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by md...@apache.org on 2023/03/11 13:16:44 UTC
[incubator-pekko-connectors-kafka] branch main updated: Remove org.apache.pekko FQCN boilerplate
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 61e9cd0c Remove org.apache.pekko FQCN boilerplate
61e9cd0c is described below
commit 61e9cd0cf43e465718ac28b112befe438ebdecc2
Author: Matthew de Detrich <ma...@aiven.io>
AuthorDate: Sat Mar 11 13:29:20 2023 +0100
Remove org.apache.pekko FQCN boilerplate
---
.../pekko/kafka/benchmarks/BatchedConsumer.scala | 7 +-
.../apache/pekko/kafka/benchmarks/Benchmarks.scala | 15 +++--
.../kafka/benchmarks/NoCommitBackpressure.scala | 5 +-
.../PekkoConnectorsCommittableProducer.scala | 7 +-
.../apache/pekko/kafka/benchmarks/Producer.scala | 12 ++--
.../pekko/kafka/benchmarks/Transactions.scala | 7 +-
.../pekko/kafka/benchmarks/InflightMetrics.scala | 11 ++--
.../kafka/benchmarks/KafkaProducerFixtureGen.scala | 5 +-
.../PekkoConnectorsCommittableSinkFixtures.scala | 21 +++---
.../ReactiveKafkaConsumerBenchmarks.scala | 19 +++---
.../benchmarks/ReactiveKafkaConsumerFixtures.scala | 15 +++--
.../ReactiveKafkaProducerBenchmarks.scala | 13 ++--
.../benchmarks/ReactiveKafkaProducerFixtures.scala | 15 +++--
.../ReactiveKafkaTransactionBenchmarks.scala | 11 ++--
.../ReactiveKafkaTransactionFixtures.scala | 19 +++---
.../org/apache/pekko/kafka/benchmarks/Timed.scala | 13 ++--
.../cluster/sharding/KafkaClusterSharding.scala | 77 +++++++++++-----------
.../org/apache/pekko/kafka/CommitterSettings.scala | 15 +++--
.../org/apache/pekko/kafka/ConsumerMessage.scala | 7 +-
.../org/apache/pekko/kafka/ConsumerSettings.scala | 31 ++++-----
.../apache/pekko/kafka/KafkaConsumerActor.scala | 9 +--
.../kafka/OffsetResetProtectionSettings.scala | 5 +-
.../org/apache/pekko/kafka/ProducerSettings.scala | 9 +--
.../apache/pekko/kafka/RestrictedConsumer.scala | 5 +-
.../org/apache/pekko/kafka/Subscriptions.scala | 27 ++++----
.../kafka/internal/BaseSingleSourceLogic.scala | 15 +++--
.../kafka/internal/CommitCollectorStage.scala | 13 ++--
.../kafka/internal/CommitObservationLogic.scala | 14 ++--
.../pekko/kafka/internal/CommittableSources.scala | 37 ++++++-----
.../internal/CommittingProducerSinkStage.scala | 19 +++---
.../pekko/kafka/internal/ConfigSettings.scala | 5 +-
.../pekko/kafka/internal/ConnectionChecker.scala | 9 +--
.../kafka/internal/ConsumerResetProtection.scala | 11 ++--
.../kafka/internal/ControlImplementations.scala | 21 +++---
.../kafka/internal/DefaultProducerStage.scala | 19 +++---
.../pekko/kafka/internal/DeferredProducer.scala | 11 ++--
.../kafka/internal/ExternalSingleSourceLogic.scala | 9 +--
.../pekko/kafka/internal/KafkaConsumerActor.scala | 23 +++----
.../pekko/kafka/internal/KafkaSourceStage.scala | 9 +--
.../pekko/kafka/internal/LoggingWithId.scala | 7 +-
.../pekko/kafka/internal/MessageBuilder.scala | 9 +--
.../internal/PartitionAssignmentHelpers.scala | 13 ++--
.../apache/pekko/kafka/internal/PlainSources.scala | 19 +++---
.../pekko/kafka/internal/ProducerStage.scala | 9 +--
.../pekko/kafka/internal/SingleSourceLogic.scala | 15 +++--
.../pekko/kafka/internal/SourceLogicBuffer.scala | 5 +-
.../kafka/internal/SourceLogicSubscription.scala | 15 +++--
.../pekko/kafka/internal/SubSourceLogic.scala | 33 +++++-----
.../internal/TransactionalProducerStage.scala | 19 +++---
.../kafka/internal/TransactionalSources.scala | 37 ++++++-----
.../org/apache/pekko/kafka/javadsl/Committer.scala | 17 ++---
.../org/apache/pekko/kafka/javadsl/Consumer.scala | 23 +++----
.../pekko/kafka/javadsl/DiscoverySupport.scala | 5 +-
.../pekko/kafka/javadsl/MetadataClient.scala | 15 +++--
.../kafka/javadsl/PartitionAssignmentHandler.scala | 11 ++--
.../org/apache/pekko/kafka/javadsl/Producer.scala | 69 +++++++++----------
.../apache/pekko/kafka/javadsl/SendProducer.scala | 15 +++--
.../apache/pekko/kafka/javadsl/Transactional.scala | 25 +++----
.../apache/pekko/kafka/scaladsl/Committer.scala | 17 ++---
.../org/apache/pekko/kafka/scaladsl/Consumer.scala | 17 ++---
.../pekko/kafka/scaladsl/DiscoverySupport.scala | 13 ++--
.../pekko/kafka/scaladsl/MetadataClient.scala | 13 ++--
.../scaladsl/PartitionAssignmentHandler.scala | 13 ++--
.../org/apache/pekko/kafka/scaladsl/Producer.scala | 69 +++++++++----------
.../apache/pekko/kafka/scaladsl/SendProducer.scala | 17 ++---
.../pekko/kafka/scaladsl/Transactional.scala | 19 +++---
.../kafka/testkit/ConsumerResultFactory.scala | 13 ++--
.../KafkaTestkitTestcontainersSettings.scala | 7 +-
.../kafka/testkit/ProducerResultFactory.scala | 7 +-
.../kafka/testkit/internal/KafkaTestKit.scala | 7 +-
.../testkit/internal/TestcontainersKafka.scala | 7 +-
.../testkit/javadsl/ConsumerControlFactory.scala | 13 ++--
.../testkit/scaladsl/ConsumerControlFactory.scala | 13 ++--
.../pekko/kafka/testkit/scaladsl/KafkaSpec.scala | 27 ++++----
.../testkit/scaladsl/TestcontainersKafkaLike.scala | 9 +--
.../org/apache/pekko/kafka/IntegrationTests.scala | 5 +-
.../kafka/PartitionedSourceFailoverSpec.scala | 13 ++--
.../pekko/kafka/PlainSourceFailoverSpec.scala | 11 ++--
.../kafka/TransactionsPartitionedSourceSpec.scala | 15 +++--
.../pekko/kafka/TransactionsSourceSpec.scala | 17 ++---
.../test/scala/docs/scaladsl/AssignmentSpec.scala | 13 ++--
.../src/test/scala/docs/scaladsl/AtLeastOnce.scala | 19 +++---
.../docs/scaladsl/ClusterShardingExample.scala | 29 ++++----
.../test/scala/docs/scaladsl/ConsumerExample.scala | 37 ++++++-----
.../test/scala/docs/scaladsl/DocsSpecBase.scala | 9 +--
.../test/scala/docs/scaladsl/FetchMetadata.scala | 14 ++--
.../scala/docs/scaladsl/PartitionExamples.scala | 19 +++---
.../test/scala/docs/scaladsl/ProducerExample.scala | 15 +++--
.../scaladsl/SchemaRegistrySerializationSpec.scala | 17 ++---
.../scala/docs/scaladsl/SendProducerSpec.scala | 13 ++--
.../scala/docs/scaladsl/SerializationSpec.scala | 17 ++---
.../scala/docs/scaladsl/TestkitSamplesSpec.scala | 22 ++++---
.../scala/docs/scaladsl/TransactionsExample.scala | 24 +++----
.../apache/pekko/kafka/ConfigSettingsSpec.scala | 5 +-
.../apache/pekko/kafka/ConsumerSettingsSpec.scala | 9 +--
.../apache/pekko/kafka/ProducerSettingsSpec.scala | 9 +--
.../org/apache/pekko/kafka/TransactionsOps.scala | 21 +++---
.../kafka/internal/CommitCollectorStageSpec.scala | 29 ++++----
.../internal/CommittingProducerSinkSpec.scala | 33 +++++-----
.../kafka/internal/CommittingWithMockSpec.scala | 25 +++----
.../kafka/internal/ConnectionCheckerSpec.scala | 13 ++--
.../apache/pekko/kafka/internal/ConsumerMock.scala | 5 +-
.../internal/ConsumerResetProtectionSpec.scala | 15 +++--
.../apache/pekko/kafka/internal/ConsumerSpec.scala | 23 +++----
.../kafka/internal/PartitionedSourceSpec.scala | 21 +++---
.../apache/pekko/kafka/internal/ProducerSpec.scala | 25 +++----
.../pekko/kafka/internal/SubscriptionsSpec.scala | 7 +-
.../apache/pekko/kafka/javadsl/ControlSpec.scala | 9 +--
.../pekko/kafka/scaladsl/CommittableSinkSpec.scala | 13 ++--
.../pekko/kafka/scaladsl/CommittingSpec.scala | 25 +++----
.../kafka/scaladsl/ConnectionCheckerSpec.scala | 13 ++--
.../apache/pekko/kafka/scaladsl/ControlSpec.scala | 7 +-
.../pekko/kafka/scaladsl/IntegrationSpec.scala | 23 +++----
.../pekko/kafka/scaladsl/MetadataClientSpec.scala | 5 +-
.../kafka/scaladsl/MisconfiguredConsumerSpec.scala | 13 ++--
.../kafka/scaladsl/MisconfiguredProducerSpec.scala | 13 ++--
.../pekko/kafka/scaladsl/MultiConsumerSpec.scala | 9 +--
.../kafka/scaladsl/PartitionedSourcesSpec.scala | 21 +++---
.../pekko/kafka/scaladsl/RebalanceExtSpec.scala | 15 +++--
.../pekko/kafka/scaladsl/RebalanceSpec.scala | 17 ++---
.../pekko/kafka/scaladsl/ReconnectSpec.scala | 11 ++--
.../pekko/kafka/scaladsl/RetentionPeriodSpec.scala | 15 +++--
.../org/apache/pekko/kafka/scaladsl/SpecBase.scala | 14 ++--
.../pekko/kafka/scaladsl/TimestampSpec.scala | 9 +--
.../pekko/kafka/scaladsl/TransactionsSpec.scala | 17 ++---
.../pekko/kafka/tests/CapturingAppender.scala | 9 +--
126 files changed, 1078 insertions(+), 966 deletions(-)
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/BatchedConsumer.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/BatchedConsumer.scala
index 07cfe610..db5f2379 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/BatchedConsumer.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/BatchedConsumer.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka.benchmarks
-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
+import org.apache.pekko
+import pekko.kafka.benchmarks.BenchmarksBase.{ topic_1000_100, topic_1000_5000, topic_1000_5000_8 }
+import pekko.kafka.benchmarks.Timed.runPerfTest
+import pekko.kafka.benchmarks.app.RunTestCommand
class ApacheKafkaBatchedConsumer extends BenchmarksBase() {
it should "bench with small messages" in {
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Benchmarks.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Benchmarks.scala
index 73a2eaf7..253a53e8 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Benchmarks.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Benchmarks.scala
@@ -14,13 +14,14 @@
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 org.apache.pekko
+import pekko.kafka.benchmarks.BenchmarksBase._
+import pekko.kafka.benchmarks.InflightMetrics._
+import pekko.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
+import pekko.kafka.benchmarks.Timed.{ runPerfTest, runPerfTestInflightMetrics }
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
import com.typesafe.config.Config
object BenchmarksBase {
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/NoCommitBackpressure.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/NoCommitBackpressure.scala
index 194e1fae..d135c5f6 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/NoCommitBackpressure.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/NoCommitBackpressure.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.kafka.benchmarks.Timed.runPerfTest
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko
+import pekko.kafka.benchmarks.Timed.runPerfTest
+import pekko.kafka.benchmarks.app.RunTestCommand
import BenchmarksBase._
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableProducer.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableProducer.scala
index 0027fb2e..7051f09c 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableProducer.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableProducer.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka.benchmarks
-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 org.apache.pekko
+import pekko.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
+import pekko.kafka.benchmarks.Timed.runPerfTest
+import pekko.kafka.benchmarks.app.RunTestCommand
/**
* Compares the `CommittingProducerSinkStage` with the composed implementation of `Producer.flexiFlow` and `Committer.sink`.
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Producer.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Producer.scala
index 76c3ebb6..25600a52 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Producer.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Producer.scala
@@ -14,14 +14,10 @@
package org.apache.pekko.kafka.benchmarks
-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
+import org.apache.pekko
+import pekko.kafka.benchmarks.BenchmarksBase.{ topic_2000_100, topic_2000_500, topic_2000_5000, topic_2000_5000_8 }
+import pekko.kafka.benchmarks.Timed.runPerfTest
+import pekko.kafka.benchmarks.app.RunTestCommand
class ApacheKafkaPlainProducer extends BenchmarksBase() {
private val prefix = "apache-kafka-plain-producer"
diff --git a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Transactions.scala b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Transactions.scala
index 82124cc9..33b1e3cc 100644
--- a/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Transactions.scala
+++ b/benchmarks/src/it/scala/org/apache/pekko/kafka/benchmarks/Transactions.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka.benchmarks
-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 org.apache.pekko
+import pekko.kafka.benchmarks.BenchmarksBase.{ topic_100_100, topic_100_5000 }
+import pekko.kafka.benchmarks.Timed.runPerfTest
+import pekko.kafka.benchmarks.app.RunTestCommand
import scala.concurrent.duration._
class ApacheKafkaTransactions extends BenchmarksBase() {
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala
index d066c755..e0aa9da9 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala
@@ -16,11 +16,12 @@ package org.apache.pekko.kafka.benchmarks
import java.lang.management.{ BufferPoolMXBean, ManagementFactory, MemoryType }
-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 org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.Cancellable
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
import com.codahale.metrics.{ Histogram, MetricRegistry }
import javax.management.remote.{ JMXConnectorFactory, JMXServiceURL }
import javax.management.{ Attribute, MBeanServerConnection, ObjectName }
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerFixtureGen.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerFixtureGen.scala
index eedfeb39..c83152c3 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerFixtureGen.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerFixtureGen.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
+import org.apache.pekko
+import pekko.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
+import pekko.kafka.benchmarks.app.RunTestCommand
import org.apache.kafka.clients.producer.KafkaProducer
case class KafkaProducerTestFixture(topic: String,
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala
index eec718cf..4f1b8db3 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala
@@ -14,16 +14,17 @@
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 org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableMessage }
+import pekko.kafka.ProducerMessage.Envelope
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
+import pekko.kafka.scaladsl.{ Committer, Consumer, Producer }
+import pekko.kafka._
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
import com.codahale.metrics.Meter
import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.clients.consumer.ConsumerConfig
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerBenchmarks.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerBenchmarks.scala
index 746213ef..88350a58 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerBenchmarks.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerBenchmarks.scala
@@ -16,15 +16,16 @@ package org.apache.pekko.kafka.benchmarks
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.ConsumerMessage.CommittableMessage
-import org.apache.pekko.kafka.benchmarks.InflightMetrics.{ BrokerMetricRequest, ConsumerMetricRequest }
-import org.apache.pekko.kafka.scaladsl.Committer
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.{ CommitDelivery, CommitterSettings }
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.ConsumerMessage.CommittableMessage
+import pekko.kafka.benchmarks.InflightMetrics.{ BrokerMetricRequest, ConsumerMetricRequest }
+import pekko.kafka.scaladsl.Committer
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.{ CommitDelivery, CommitterSettings }
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
import com.codahale.metrics.Meter
import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.clients.consumer.ConsumerRecord
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerFixtures.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerFixtures.scala
index 2cfcb98a..2e047640 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerFixtures.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerFixtures.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage.CommittableMessage
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.{ ConsumerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl.Source
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.CommittableMessage
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.kafka.scaladsl.Consumer
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.{ ConsumerSettings, Subscriptions }
+import pekko.stream.scaladsl.Source
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord }
import org.apache.kafka.common.serialization.{ ByteArrayDeserializer, StringDeserializer }
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerBenchmarks.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerBenchmarks.scala
index 976f3148..c9e8fcfd 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerBenchmarks.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerBenchmarks.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.kafka.ConsumerMessage.CommittableMessage
-import org.apache.pekko.kafka.ProducerMessage
-import org.apache.pekko.kafka.ProducerMessage.{ Result, Results }
-import org.apache.pekko.kafka.benchmarks.ReactiveKafkaProducerFixtures.ReactiveKafkaProducerTestFixture
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.{ Sink, Source }
+import org.apache.pekko
+import pekko.kafka.ConsumerMessage.CommittableMessage
+import pekko.kafka.ProducerMessage
+import pekko.kafka.ProducerMessage.{ Result, Results }
+import pekko.kafka.benchmarks.ReactiveKafkaProducerFixtures.ReactiveKafkaProducerTestFixture
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Sink, Source }
import com.codahale.metrics.Meter
import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.clients.producer.ProducerRecord
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerFixtures.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerFixtures.scala
index e38f804b..85d2e3f9 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerFixtures.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerFixtures.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.NotUsed
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ProducerMessage.{ Envelope, Results }
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
-import org.apache.pekko.kafka.scaladsl.Producer
-import org.apache.pekko.stream.scaladsl.Flow
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.ActorSystem
+import pekko.kafka.ProducerMessage.{ Envelope, Results }
+import pekko.kafka.ProducerSettings
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.kafka.scaladsl.Producer
+import pekko.stream.scaladsl.Flow
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.{ ByteArraySerializer, StringSerializer }
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala
index 537b987d..8564bf08 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.kafka.ProducerMessage
-import org.apache.pekko.kafka.ProducerMessage.{ Result, Results }
-import org.apache.pekko.kafka.benchmarks.ReactiveKafkaTransactionFixtures._
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink }
+import org.apache.pekko
+import pekko.kafka.ProducerMessage
+import pekko.kafka.ProducerMessage.{ Result, Results }
+import pekko.kafka.benchmarks.ReactiveKafkaTransactionFixtures._
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Keep, Sink }
import com.codahale.metrics.Meter
import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.clients.producer.ProducerRecord
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionFixtures.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionFixtures.scala
index 79b9d6c6..0f6f7274 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionFixtures.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionFixtures.scala
@@ -14,15 +14,16 @@
package org.apache.pekko.kafka.benchmarks
-import org.apache.pekko.NotUsed
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage.TransactionalMessage
-import org.apache.pekko.kafka.ProducerMessage.{ Envelope, Results }
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.Transactional
-import org.apache.pekko.kafka.{ ConsumerMessage, ConsumerSettings, ProducerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl.{ Flow, Source }
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.TransactionalMessage
+import pekko.kafka.ProducerMessage.{ Envelope, Results }
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.Transactional
+import pekko.kafka.{ ConsumerMessage, ConsumerSettings, ProducerSettings, Subscriptions }
+import pekko.stream.scaladsl.{ Flow, Source }
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.{
ByteArrayDeserializer,
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/Timed.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/Timed.scala
index 143579d6..ce8cc1f0 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/Timed.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/Timed.scala
@@ -16,12 +16,13 @@ package org.apache.pekko.kafka.benchmarks
import com.codahale.metrics._
import com.typesafe.scalalogging.LazyLogging
-import org.apache.pekko.NotUsed
-import org.apache.pekko.kafka.benchmarks.InflightMetrics.{ BrokerMetricRequest, ConsumerMetricRequest }
-import org.apache.pekko.kafka.benchmarks.app.RunTestCommand
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.{ FileIO, Flow, Sink, Source }
-import org.apache.pekko.util.ByteString
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.kafka.benchmarks.InflightMetrics.{ BrokerMetricRequest, ConsumerMetricRequest }
+import pekko.kafka.benchmarks.app.RunTestCommand
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ FileIO, Flow, Sink, Source }
+import pekko.util.ByteString
import java.nio.charset.{ Charset, StandardCharsets }
import java.nio.file.Paths
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala b/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
index ab63f3c5..56545cc8 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
@@ -17,24 +17,25 @@ package org.apache.pekko.kafka.cluster.sharding
import java.util.concurrent.{ CompletionStage, ConcurrentHashMap }
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.actor.typed.Behavior
-import org.apache.pekko.actor.typed.scaladsl.Behaviors
-import org.apache.pekko.actor.typed.scaladsl.adapter._
-import org.apache.pekko.actor.{ ActorSystem, ClassicActorSystemProvider, ExtendedActorSystem, Extension, ExtensionId }
-import org.apache.pekko.annotation.{ ApiMayChange, InternalApi }
-import org.apache.pekko.cluster.sharding.external.ExternalShardAllocation
-import org.apache.pekko.cluster.sharding.typed.scaladsl.EntityTypeKey
-import org.apache.pekko.cluster.sharding.typed.{ ShardingEnvelope, ShardingMessageExtractor }
-import org.apache.pekko.cluster.typed.Cluster
-import org.apache.pekko.kafka.scaladsl.MetadataClient
-import org.apache.pekko.kafka._
-import org.apache.pekko.util.Timeout._
+import org.apache.pekko
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.actor.{ ActorSystem, ClassicActorSystemProvider, ExtendedActorSystem, Extension, ExtensionId }
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.cluster.sharding.external.ExternalShardAllocation
+import pekko.cluster.sharding.typed.scaladsl.EntityTypeKey
+import pekko.cluster.sharding.typed.{ ShardingEnvelope, ShardingMessageExtractor }
+import pekko.cluster.typed.Cluster
+import pekko.kafka.scaladsl.MetadataClient
+import pekko.kafka._
+import pekko.util.Timeout._
import org.apache.kafka.common.utils.Utils
import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContextExecutor, Future }
import scala.util.{ Failure, Success }
-import org.apache.pekko.util.JavaDurationConverters._
+import pekko.util.JavaDurationConverters._
import org.slf4j.LoggerFactory
import scala.compat.java8.FutureConverters._
@@ -51,7 +52,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
/**
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka
@@ -59,7 +60,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* the Kafka Consumer connection required to retrieve the number of partitions. Each call to this method will result
* in a round trip to Kafka. This method should only be called once per entity type [[M]], per local actor system.
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -73,7 +74,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
*
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka
@@ -81,7 +82,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* the Kafka Consumer connection required to retrieve the number of partitions. Each call to this method will result
* in a round trip to Kafka. This method should only be called once per entity type [[M]], per local actor system.
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -95,12 +96,12 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
/**
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy is provided explicitly with [[kafkaPartitions]].
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -110,7 +111,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
/**
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka
@@ -119,7 +120,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* a field from the Entity to use as the entity id for the hashing strategy. Each call to this method will result
* in a round trip to Kafka. This method should only be called once per entity type [[M]], per local actor system.
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -135,7 +136,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
*
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy will be automatically determined by querying the Kafka
@@ -144,7 +145,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* a field from the Entity to use as the entity id for the hashing strategy. Each call to this method will result
* in a round trip to Kafka. This method should only be called once per entity type [[M]], per local actor system.
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -161,12 +162,12 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
/**
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy is provided explicitly with [[kafkaPartitions]].
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -177,12 +178,12 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
/**
* API MAY CHANGE
*
- * Asynchronously return a [[org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
+ * Asynchronously return a [[pekko.cluster.sharding.typed.ShardingMessageExtractor]] with a default hashing strategy
* based on Apache Kafka's [[org.apache.kafka.clients.producer.internals.DefaultPartitioner]].
*
* The number of partitions to use with the hashing strategy is provided explicitly with [[kafkaPartitions]].
*
- * All topics used in a Consumer [[org.apache.pekko.kafka.Subscription]] must contain the same number of partitions to ensure
+ * All topics used in a Consumer [[pekko.kafka.Subscription]] must contain the same number of partitions to ensure
* that entities are routed to the same Entity type.
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
@@ -209,7 +210,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
}
private val rebalanceListeners =
- new ConcurrentHashMap[EntityTypeKey[_], org.apache.pekko.actor.typed.ActorRef[ConsumerRebalanceEvent]]()
+ new ConcurrentHashMap[EntityTypeKey[_], pekko.actor.typed.ActorRef[ConsumerRebalanceEvent]]()
/**
* API MAY CHANGE
@@ -219,16 +220,17 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* the rebalance listener will use the [[ExternalShardAllocation]] client to update the External Sharding strategy
* accordingly so that entities are (eventually) routed to the local Apache Pekko cluster member.
*
- * Returns an Apache Pekko typed [[org.apache.pekko.actor.typed.ActorRef]]. This must be converted to a classic actor before it can be
+ * Returns an Apache Pekko typed [[pekko.actor.typed.ActorRef]]. This must be converted to a classic actor before it can be
* passed to an Apache Pekko Connector Kafka [[ConsumerSettings]].
*
* {{{
- * import org.apache.pekko.actor.typed.scaladsl.adapter._
- * val listenerClassicActorRef: org.apache.pekko.actor.ActorRef = listenerTypedActorRef.toClassic
+ * import org.apache.pekko
+ * import pekko.actor.typed.scaladsl.adapter._
+ * val listenerClassicActorRef: pekko.actor.ActorRef = listenerTypedActorRef.toClassic
* }}}
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
- def rebalanceListener(typeKey: EntityTypeKey[_]): org.apache.pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] = {
+ def rebalanceListener(typeKey: EntityTypeKey[_]): pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] = {
rebalanceListeners.computeIfAbsent(typeKey,
_ => {
system.toTyped
@@ -246,18 +248,19 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
* the rebalance listener will use the [[ExternalShardAllocation]] client to update the External Sharding strategy
* accordingly so that entities are (eventually) routed to the local Apache Pekko cluster member.
*
- * Returns an Apache Pekko typed [[org.apache.pekko.actor.typed.ActorRef]]. This must be converted to a classic actor before it can be
+ * Returns an Apache Pekko typed [[pekko.actor.typed.ActorRef]]. This must be converted to a classic actor before it can be
* passed to an Apache Pekko Connector Kafka [[ConsumerSettings]].
*
* {{{
- * import org.apache.pekko.actor.typed.scaladsl.adapter._
- * val listenerClassicActorRef: org.apache.pekko.actor.ActorRef = listenerTypedActorRef.toClassic
+ * import org.apache.pekko
+ * import pekko.actor.typed.scaladsl.adapter._
+ * val listenerClassicActorRef: pekko.actor.ActorRef = listenerTypedActorRef.toClassic
* }}}
*/
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/1074")
def rebalanceListener(
- typeKey: org.apache.pekko.cluster.sharding.typed.javadsl.EntityTypeKey[_])
- : org.apache.pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] = {
+ typeKey: pekko.cluster.sharding.typed.javadsl.EntityTypeKey[_])
+ : pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] = {
rebalanceListener(typeKey.asScala)
}
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala b/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
index 7d069426..d1c75b9d 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
@@ -15,8 +15,9 @@
package org.apache.pekko.kafka
import java.util.concurrent.TimeUnit
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.util.JavaDurationConverters._
import com.typesafe.config.Config
import scala.concurrent.duration._
@@ -113,7 +114,7 @@ object CommitterSettings {
* Create settings from the default configuration
* `pekko.kafka.committer`.
*/
- def apply(actorSystem: org.apache.pekko.actor.ActorSystem): CommitterSettings =
+ def apply(actorSystem: pekko.actor.ActorSystem): CommitterSettings =
apply(actorSystem.settings.config.getConfig(configPath))
/**
@@ -122,7 +123,7 @@ object CommitterSettings {
*
* For use with the `pekko.actor.typed` API.
*/
- def apply(actorSystem: org.apache.pekko.actor.ClassicActorSystemProvider): CommitterSettings =
+ def apply(actorSystem: pekko.actor.ClassicActorSystemProvider): CommitterSettings =
apply(actorSystem.classicSystem.settings.config.getConfig(configPath))
/**
@@ -142,7 +143,7 @@ object CommitterSettings {
* Java API: Create settings from the default configuration
* `pekko.kafka.committer`.
*/
- def create(actorSystem: org.apache.pekko.actor.ActorSystem): CommitterSettings =
+ def create(actorSystem: pekko.actor.ActorSystem): CommitterSettings =
apply(actorSystem)
/**
@@ -151,7 +152,7 @@ object CommitterSettings {
*
* For use with the `pekko.actor.typed` API.
*/
- def create(actorSystem: org.apache.pekko.actor.ClassicActorSystemProvider): CommitterSettings =
+ def create(actorSystem: pekko.actor.ClassicActorSystemProvider): CommitterSettings =
apply(actorSystem)
/**
@@ -165,7 +166,7 @@ object CommitterSettings {
/**
* Settings for committer. See `pekko.kafka.committer` section in
- * reference.conf. Note that the [[org.apache.pekko.kafka.CommitterSettings$ companion]] object provides
+ * reference.conf. Note that the [[pekko.kafka.CommitterSettings companion]] object provides
* `apply` and `create` functions for convenient construction of the settings, together with
* the `with` methods.
*/
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala b/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
index 9cfb47fb..1be1bc28 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
@@ -17,9 +17,10 @@ package org.apache.pekko.kafka
import java.util.Objects
import java.util.concurrent.CompletionStage
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.{ DoNotInherit, InternalApi }
-import org.apache.pekko.kafka.internal.{ CommittableOffsetBatchImpl, CommittedMarker }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.{ DoNotInherit, InternalApi }
+import pekko.kafka.internal.{ CommittableOffsetBatchImpl, CommittedMarker }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala b/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
index 07330b35..bf1131ad 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
@@ -17,9 +17,10 @@ package org.apache.pekko.kafka
import java.util.Optional
import java.util.concurrent.{ CompletionStage, Executor }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.internal._
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.kafka.internal._
+import pekko.util.JavaDurationConverters._
import com.typesafe.config.Config
import org.apache.kafka.clients.consumer.{ Consumer, ConsumerConfig, KafkaConsumer }
import org.apache.kafka.common.serialization.Deserializer
@@ -40,7 +41,7 @@ object ConsumerSettings {
* Key or value deserializer can be passed explicitly or retrieved from configuration.
*/
def apply[K, V](
- system: org.apache.pekko.actor.ActorSystem,
+ system: pekko.actor.ActorSystem,
keyDeserializer: Option[Deserializer[K]],
valueDeserializer: Option[Deserializer[V]]): ConsumerSettings[K, V] = {
val config = system.settings.config.getConfig(configPath)
@@ -55,7 +56,7 @@ object ConsumerSettings {
* For use with the `pekko.actor.typed` API.
*/
def apply[K, V](
- system: org.apache.pekko.actor.ClassicActorSystemProvider,
+ system: pekko.actor.ClassicActorSystemProvider,
keyDeserializer: Option[Deserializer[K]],
valueDeserializer: Option[Deserializer[V]]): ConsumerSettings[K, V] =
apply(system.classicSystem, keyDeserializer, valueDeserializer)
@@ -126,7 +127,7 @@ object ConsumerSettings {
* Key and value serializer must be passed explicitly.
*/
def apply[K, V](
- system: org.apache.pekko.actor.ActorSystem,
+ system: pekko.actor.ActorSystem,
keyDeserializer: Deserializer[K],
valueDeserializer: Deserializer[V]): ConsumerSettings[K, V] =
apply(system, Option(keyDeserializer), Option(valueDeserializer))
@@ -139,7 +140,7 @@ object ConsumerSettings {
* For use with the `pekko.actor.typed` API.
*/
def apply[K, V](
- system: org.apache.pekko.actor.ClassicActorSystemProvider,
+ system: pekko.actor.ClassicActorSystemProvider,
keyDeserializer: Deserializer[K],
valueDeserializer: Deserializer[V]): ConsumerSettings[K, V] =
apply(system, Option(keyDeserializer), Option(valueDeserializer))
@@ -161,7 +162,7 @@ object ConsumerSettings {
* Key or value deserializer can be passed explicitly or retrieved from configuration.
*/
def create[K, V](
- system: org.apache.pekko.actor.ActorSystem,
+ system: pekko.actor.ActorSystem,
keyDeserializer: Optional[Deserializer[K]],
valueDeserializer: Optional[Deserializer[V]]): ConsumerSettings[K, V] =
apply(system, keyDeserializer.asScala, valueDeserializer.asScala)
@@ -174,7 +175,7 @@ object ConsumerSettings {
* For use with the `pekko.actor.typed` API.
*/
def create[K, V](
- system: org.apache.pekko.actor.ClassicActorSystemProvider,
+ system: pekko.actor.ClassicActorSystemProvider,
keyDeserializer: Optional[Deserializer[K]],
valueDeserializer: Optional[Deserializer[V]]): ConsumerSettings[K, V] =
apply(system, keyDeserializer.asScala, valueDeserializer.asScala)
@@ -196,7 +197,7 @@ object ConsumerSettings {
* Key and value serializer must be passed explicitly.
*/
def create[K, V](
- system: org.apache.pekko.actor.ActorSystem,
+ system: pekko.actor.ActorSystem,
keyDeserializer: Deserializer[K],
valueDeserializer: Deserializer[V]): ConsumerSettings[K, V] =
apply(system, keyDeserializer, valueDeserializer)
@@ -209,7 +210,7 @@ object ConsumerSettings {
* For use with the `pekko.actor.typed` API.
*/
def create[K, V](
- system: org.apache.pekko.actor.ClassicActorSystemProvider,
+ system: pekko.actor.ClassicActorSystemProvider,
keyDeserializer: Deserializer[K],
valueDeserializer: Deserializer[V]): ConsumerSettings[K, V] =
apply(system, keyDeserializer, valueDeserializer)
@@ -237,7 +238,7 @@ object ConsumerSettings {
/**
* Settings for consumers. See `pekko.kafka.consumer` section in
- * `reference.conf`. Note that the [[org.apache.pekko.kafka.ConsumerSettings$ companion]] object provides
+ * `reference.conf`. Note that the [[pekko.kafka.ConsumerSettings companion]] object provides
* `apply` and `create` functions for convenient construction of the settings, together with
* the `with` methods.
*
@@ -385,7 +386,7 @@ class ConsumerSettings[K, V] @InternalApi private[kafka] (
/**
* If offset commit requests are not completed within this timeout
- * the returned Future is completed with [[org.apache.pekko.kafka.CommitTimeoutException]].
+ * the returned Future is completed with [[pekko.kafka.CommitTimeoutException]].
*/
def withCommitTimeout(commitTimeout: FiniteDuration): ConsumerSettings[K, V] =
copy(commitTimeout = commitTimeout)
@@ -393,7 +394,7 @@ class ConsumerSettings[K, V] @InternalApi private[kafka] (
/**
* Java API:
* If offset commit requests are not completed within this timeout
- * the returned Future is completed with [[org.apache.pekko.kafka.CommitTimeoutException]].
+ * the returned Future is completed with [[pekko.kafka.CommitTimeoutException]].
*/
def withCommitTimeout(commitTimeout: java.time.Duration): ConsumerSettings[K, V] =
copy(commitTimeout = commitTimeout.asScala)
@@ -413,7 +414,7 @@ class ConsumerSettings[K, V] @InternalApi private[kafka] (
/**
* Fully qualified config path which holds the dispatcher configuration
- * to be used by the [[org.apache.pekko.kafka.KafkaConsumerActor]]. Some blocking may occur.
+ * to be used by the [[pekko.kafka.KafkaConsumerActor]]. Some blocking may occur.
*/
def withDispatcher(dispatcher: String): ConsumerSettings[K, V] =
copy(dispatcher = dispatcher)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/KafkaConsumerActor.scala b/core/src/main/scala/org/apache/pekko/kafka/KafkaConsumerActor.scala
index 47004959..457a5465 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/KafkaConsumerActor.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/KafkaConsumerActor.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka
-import org.apache.pekko.actor.{ ActorRef, NoSerializationVerificationNeeded, Props }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.internal.{ KafkaConsumerActor => InternalKafkaConsumerActor }
+import org.apache.pekko
+import pekko.actor.{ ActorRef, NoSerializationVerificationNeeded, Props }
+import pekko.annotation.InternalApi
+import pekko.kafka.internal.{ KafkaConsumerActor => InternalKafkaConsumerActor }
object KafkaConsumerActor {
@@ -44,7 +45,7 @@ object KafkaConsumerActor {
/**
* Creates Props for the Kafka Consumer Actor with a reference back to the owner of it
- * which will be signalled with [[org.apache.pekko.actor.Status.Failure Failure(exception)]], in case the
+ * which will be signalled with [[pekko.actor.Status.Failure Failure(exception)]], in case the
* Kafka client instance can't be created.
*/
def props[K, V](owner: ActorRef, settings: ConsumerSettings[K, V]): Props =
diff --git a/core/src/main/scala/org/apache/pekko/kafka/OffsetResetProtectionSettings.scala b/core/src/main/scala/org/apache/pekko/kafka/OffsetResetProtectionSettings.scala
index 306b5af2..d90f62f0 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/OffsetResetProtectionSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/OffsetResetProtectionSettings.scala
@@ -15,8 +15,9 @@
package org.apache.pekko.kafka
import java.time.{ Duration => JDuration }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.util.JavaDurationConverters._
import com.typesafe.config.Config
import scala.concurrent.duration._
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
index 07e325e0..3b1cc151 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
@@ -17,8 +17,9 @@ package org.apache.pekko.kafka
import java.util.Optional
import java.util.concurrent.{ CompletionStage, Executor }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.internal.ConfigSettings
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.kafka.internal.ConfigSettings
import com.typesafe.config.Config
import org.apache.kafka.clients.producer.{ KafkaProducer, Producer, ProducerConfig }
import org.apache.kafka.common.serialization.Serializer
@@ -26,7 +27,7 @@ import org.apache.kafka.common.serialization.Serializer
import scala.jdk.CollectionConverters._
import scala.compat.java8.OptionConverters._
import scala.concurrent.duration._
-import org.apache.pekko.util.JavaDurationConverters._
+import pekko.util.JavaDurationConverters._
import scala.concurrent.{ ExecutionContext, Future }
import scala.compat.java8.FutureConverters._
@@ -211,7 +212,7 @@ object ProducerSettings {
/**
* Settings for producers. See `pekko.kafka.producer` section in
- * reference.conf. Note that the [[org.apache.pekko.kafka.ProducerSettings$ companion]] object provides
+ * reference.conf. Note that the [[pekko.kafka.ProducerSettings companion]] object provides
* `apply` and `create` functions for convenient construction of the settings, together with
* the `with` methods.
*
diff --git a/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala b/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
index f86a677d..00c6d8a9 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka
-import org.apache.pekko.annotation.ApiMayChange
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
import org.apache.kafka.clients.consumer.{ Consumer, OffsetAndMetadata, OffsetAndTimestamp }
import org.apache.kafka.common.TopicPartition
/**
* Offers parts of the [[org.apache.kafka.clients.consumer.Consumer]] API which becomes available to
- * the [[org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler]] callbacks.
+ * the [[pekko.kafka.scaladsl.PartitionAssignmentHandler]] callbacks.
*/
@ApiMayChange
final class RestrictedConsumer(consumer: Consumer[_, _], duration: java.time.Duration) {
diff --git a/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala b/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
index 86c63654..b166c406 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.{ ApiMayChange, InternalApi }
-import org.apache.pekko.kafka.internal.PartitionAssignmentHelpers
-import org.apache.pekko.kafka.internal.PartitionAssignmentHelpers.EmptyPartitionAssignmentHandler
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.kafka.internal.PartitionAssignmentHelpers
+import pekko.kafka.internal.PartitionAssignmentHelpers.EmptyPartitionAssignmentHandler
import org.apache.kafka.common.TopicPartition
import scala.annotation.varargs
@@ -25,10 +26,10 @@ import scala.jdk.CollectionConverters._
sealed trait Subscription {
- /** ActorRef which is to receive [[org.apache.pekko.kafka.ConsumerRebalanceEvent]] signals when rebalancing happens */
+ /** ActorRef which is to receive [[pekko.kafka.ConsumerRebalanceEvent]] signals when rebalancing happens */
def rebalanceListener: Option[ActorRef]
- /** Configure this actor ref to receive [[org.apache.pekko.kafka.ConsumerRebalanceEvent]] signals */
+ /** Configure this actor ref to receive [[pekko.kafka.ConsumerRebalanceEvent]] signals */
def withRebalanceListener(ref: ActorRef): Subscription
def renderStageAttribute: String
@@ -55,13 +56,13 @@ sealed trait ManualSubscription extends Subscription {
*/
sealed trait AutoSubscription extends Subscription {
- /** ActorRef which is to receive [[org.apache.pekko.kafka.ConsumerRebalanceEvent]] signals when rebalancing happens */
+ /** ActorRef which is to receive [[pekko.kafka.ConsumerRebalanceEvent]] signals when rebalancing happens */
def rebalanceListener: Option[ActorRef]
@InternalApi
def partitionAssignmentHandler: scaladsl.PartitionAssignmentHandler
- /** Configure this actor ref to receive [[org.apache.pekko.kafka.ConsumerRebalanceEvent]] signals */
+ /** Configure this actor ref to receive [[pekko.kafka.ConsumerRebalanceEvent]] signals */
def withRebalanceListener(ref: ActorRef): AutoSubscription
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/985")
@@ -89,7 +90,7 @@ final case class TopicPartitionsRevoked(sub: Subscription, topicPartitions: Set[
object Subscriptions {
/** INTERNAL API */
- @org.apache.pekko.annotation.InternalApi
+ @pekko.annotation.InternalApi
private[kafka] final case class TopicSubscription(
tps: Set[String],
rebalanceListener: Option[ActorRef],
@@ -110,7 +111,7 @@ object Subscriptions {
}
/** INTERNAL API */
- @org.apache.pekko.annotation.InternalApi
+ @pekko.annotation.InternalApi
private[kafka] final case class TopicSubscriptionPattern(
pattern: String,
rebalanceListener: Option[ActorRef],
@@ -130,14 +131,14 @@ object Subscriptions {
}
/** INTERNAL API */
- @org.apache.pekko.annotation.InternalApi
+ @pekko.annotation.InternalApi
private[kafka] final case class Assignment(tps: Set[TopicPartition]) extends ManualSubscription {
def withRebalanceListener(ref: ActorRef): Assignment = this
def renderStageAttribute: String = s"${tps.mkString(" ")}"
}
/** INTERNAL API */
- @org.apache.pekko.annotation.InternalApi
+ @pekko.annotation.InternalApi
private[kafka] final case class AssignmentWithOffset(tps: Map[TopicPartition, Long]) extends ManualSubscription {
def withRebalanceListener(ref: ActorRef): AssignmentWithOffset = this
def renderStageAttribute: String =
@@ -145,7 +146,7 @@ object Subscriptions {
}
/** INTERNAL API */
- @org.apache.pekko.annotation.InternalApi
+ @pekko.annotation.InternalApi
private[kafka] final case class AssignmentOffsetsForTimes(timestampsToSearch: Map[TopicPartition, Long])
extends ManualSubscription {
def withRebalanceListener(ref: ActorRef): AssignmentOffsetsForTimes = this
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala
index 03f512ba..a5149052 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.{ ActorRef, Status, Terminated }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.Subscriptions.{ Assignment, AssignmentOffsetsForTimes, AssignmentWithOffset }
-import org.apache.pekko.kafka.{ ConsumerFailed, ManualSubscription }
-import org.apache.pekko.stream.SourceShape
-import org.apache.pekko.stream.stage.GraphStageLogic.StageActor
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic, OutHandler }
+import org.apache.pekko
+import pekko.actor.{ ActorRef, Status, Terminated }
+import pekko.annotation.InternalApi
+import pekko.kafka.Subscriptions.{ Assignment, AssignmentOffsetsForTimes, AssignmentWithOffset }
+import pekko.kafka.{ ConsumerFailed, ManualSubscription }
+import pekko.stream.SourceShape
+import pekko.stream.stage.GraphStageLogic.StageActor
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic, OutHandler }
import org.apache.kafka.common.TopicPartition
import scala.annotation.tailrec
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/CommitCollectorStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/CommitCollectorStage.scala
index 4c4abe18..43a4cc18 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/CommitCollectorStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/CommitCollectorStage.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.CommitterSettings
-import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.stage._
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.kafka.CommitterSettings
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
+import pekko.stream._
+import pekko.stream.stage._
/**
* INTERNAL API.
@@ -53,7 +54,7 @@ private final class CommitCollectorStageLogic(
// Context propagation is needed to notify Lightbend Telemetry to keep the context in case of a deferred downstream
// push call that might not happen during onPush but later onTimer, onPull, or only during the next onPush call.
- private val contextPropagation = org.apache.pekko.stream.impl.ContextPropagation()
+ private val contextPropagation = pekko.stream.impl.ContextPropagation()
private var contextSuspended = false
override protected def logSource: Class[_] = classOf[CommitCollectorStageLogic]
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala
index 5eef8d0c..d73e7ee8 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala
@@ -14,15 +14,11 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.kafka.CommitWhen.OffsetFirstObserved
-import org.apache.pekko.kafka.CommitterSettings
-import org.apache.pekko.kafka.ConsumerMessage.{
- Committable,
- CommittableOffset,
- CommittableOffsetBatch,
- GroupTopicPartition
-}
-import org.apache.pekko.stream.stage.GraphStageLogic
+import org.apache.pekko
+import pekko.kafka.CommitWhen.OffsetFirstObserved
+import pekko.kafka.CommitterSettings
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffset, CommittableOffsetBatch, GroupTopicPartition }
+import pekko.stream.stage.GraphStageLogic
/**
* Shared commit observation logic between [[GraphStageLogic]] that facilitate offset commits,
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
index 668c6938..1dfeb93d 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
@@ -14,20 +14,21 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.{ Commit, CommitSingle, CommitWithoutReply }
-import org.apache.pekko.kafka.internal.SubSourceLogic._
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.pattern.AskTimeoutException
-import org.apache.pekko.stream.SourceShape
-import org.apache.pekko.stream.scaladsl.Source
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
-import org.apache.pekko.util.Timeout
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
+import pekko.kafka._
+import pekko.kafka.internal.KafkaConsumerActor.Internal.{ Commit, CommitSingle, CommitWithoutReply }
+import pekko.kafka.internal.SubSourceLogic._
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.pattern.AskTimeoutException
+import pekko.stream.SourceShape
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
+import pekko.util.Timeout
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord, OffsetAndMetadata }
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.requests.OffsetFetchResponse
@@ -186,9 +187,9 @@ private[kafka] object KafkaAsyncConsumerCommitterRef {
/**
* Internal API.
*
- * Sends [[org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.Commit]],
- * [[org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.CommitSingle]] and
- * [[org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.CommitWithoutReply]] messages to the consumer actor.
+ * Sends [[pekko.kafka.internal.KafkaConsumerActor.Internal.Commit]],
+ * [[pekko.kafka.internal.KafkaConsumerActor.Internal.CommitSingle]] and
+ * [[pekko.kafka.internal.KafkaConsumerActor.Internal.CommitWithoutReply]] messages to the consumer actor.
*/
@InternalApi
private[kafka] class KafkaAsyncConsumerCommitterRef(private val consumerActor: ActorRef,
@@ -207,7 +208,7 @@ private[kafka] class KafkaAsyncConsumerCommitterRef(private val consumerActor: A
}
private def sendWithReply(msg: AnyRef): Future[Done] = {
- import org.apache.pekko.pattern.ask
+ import pekko.pattern.ask
consumerActor
.ask(msg)(Timeout(commitTimeout))
.map(_ => Done)(ExecutionContexts.parasitic)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala
index 526eca00..799eb2b2 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala
@@ -16,15 +16,16 @@ package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.{ CommitDelivery, CommitterSettings, ProducerSettings }
-import org.apache.pekko.stream.ActorAttributes.SupervisionStrategy
-import org.apache.pekko.stream.Supervision.Decider
-import org.apache.pekko.stream.stage._
-import org.apache.pekko.stream.{ Attributes, Inlet, SinkShape, Supervision }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.InternalApi
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.{ CommitDelivery, CommitterSettings, ProducerSettings }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.Supervision.Decider
+import pekko.stream.stage._
+import pekko.stream.{ Attributes, Inlet, SinkShape, Supervision }
import org.apache.kafka.clients.producer.{ Callback, RecordMetadata }
import scala.concurrent.{ Future, Promise }
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ConfigSettings.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ConfigSettings.scala
index 1b69d0db..3635ff30 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ConfigSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ConfigSettings.scala
@@ -16,13 +16,14 @@ package org.apache.pekko.kafka.internal
import java.util
-import org.apache.pekko.annotation.InternalApi
+import org.apache.pekko
+import pekko.annotation.InternalApi
import com.typesafe.config.{ Config, ConfigObject }
import scala.annotation.tailrec
import scala.jdk.CollectionConverters._
import scala.concurrent.duration.Duration
-import org.apache.pekko.util.JavaDurationConverters._
+import pekko.util.JavaDurationConverters._
/**
* INTERNAL API
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ConnectionChecker.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ConnectionChecker.scala
index 8d4ff3f4..9202a192 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ConnectionChecker.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ConnectionChecker.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.{ Actor, ActorLogging, Props, Timers }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.event.LoggingReceive
-import org.apache.pekko.kafka.{ ConnectionCheckerSettings, KafkaConnectionFailed, Metadata }
+import org.apache.pekko
+import pekko.actor.{ Actor, ActorLogging, Props, Timers }
+import pekko.annotation.InternalApi
+import pekko.event.LoggingReceive
+import pekko.kafka.{ ConnectionCheckerSettings, KafkaConnectionFailed, Metadata }
import org.apache.kafka.common.errors.TimeoutException
import scala.concurrent.duration.FiniteDuration
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerResetProtection.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerResetProtection.scala
index 3b27dbea..e955f400 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerResetProtection.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerResetProtection.scala
@@ -16,11 +16,12 @@ package org.apache.pekko.kafka.internal
import java.util
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.kafka.OffsetResetProtectionSettings
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.Seek
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.event.LoggingAdapter
+import pekko.kafka.OffsetResetProtectionSettings
+import pekko.kafka.internal.KafkaConsumerActor.Internal.Seek
import org.apache.kafka.clients.consumer.{ ConsumerRecord, ConsumerRecords, OffsetAndMetadata }
import org.apache.kafka.common.TopicPartition
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
index cc21ab56..69f73773 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
@@ -15,15 +15,16 @@
package org.apache.pekko.kafka.internal
import java.util.concurrent.{ CompletionStage, Executor }
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.{ ConsumerMetrics, RequestMetrics }
-import org.apache.pekko.kafka.{ javadsl, scaladsl }
-import org.apache.pekko.stream.SourceShape
-import org.apache.pekko.stream.stage.GraphStageLogic
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.internal.KafkaConsumerActor.Internal.{ ConsumerMetrics, RequestMetrics }
+import pekko.kafka.{ javadsl, scaladsl }
+import pekko.stream.SourceShape
+import pekko.stream.stage.GraphStageLogic
+import pekko.util.Timeout
import org.apache.kafka.common.{ Metric, MetricName }
import scala.jdk.CollectionConverters._
@@ -87,7 +88,7 @@ private trait MetricsControl extends scaladsl.Consumer.Control {
// FIXME: this can't be accessed until the stream has materialized because the `def executionContext` implementation
// takes the executioncontext from the materializer. should it throw an exception, or block, until materialization?
def metrics: Future[Map[MetricName, Metric]] = {
- import org.apache.pekko.pattern.ask
+ import pekko.pattern.ask
import scala.concurrent.duration._
consumerFuture
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala
index dd902b77..b298a4fd 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala
@@ -14,15 +14,16 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.kafka.internal.ProducerStage.ProducerCompletionState
-import org.apache.pekko.stream.ActorAttributes.SupervisionStrategy
-import org.apache.pekko.stream.Supervision.Decider
-import org.apache.pekko.stream.stage._
-import org.apache.pekko.stream.{ Attributes, FlowShape, Supervision }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.InternalApi
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.ProducerSettings
+import pekko.kafka.internal.ProducerStage.ProducerCompletionState
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.Supervision.Decider
+import pekko.stream.stage._
+import pekko.stream.{ Attributes, FlowShape, Supervision }
import org.apache.kafka.clients.producer.{ Callback, ProducerRecord, RecordMetadata }
import scala.concurrent.{ ExecutionContext, Future, Promise }
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/DeferredProducer.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/DeferredProducer.scala
index b2c26797..3c1d084a 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/DeferredProducer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/DeferredProducer.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.stream.stage._
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.ProducerSettings
+import pekko.stream.stage._
+import pekko.util.JavaDurationConverters._
import org.apache.kafka.clients.producer.Producer
import scala.util.control.NonFatal
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ExternalSingleSourceLogic.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ExternalSingleSourceLogic.scala
index 03080041..90d28153 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ExternalSingleSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ExternalSingleSourceLogic.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ManualSubscription
-import org.apache.pekko.stream.SourceShape
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.kafka.ManualSubscription
+import pekko.stream.SourceShape
import scala.concurrent.Future
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
index 679f748f..cad055a3 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
@@ -17,9 +17,10 @@ package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.locks.LockSupport
import java.util.regex.Pattern
-import org.apache.pekko.Done
-import org.apache.pekko.actor.Status.Failure
-import org.apache.pekko.actor.{
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.Status.Failure
+import pekko.actor.{
Actor,
ActorRef,
DeadLetterSuppression,
@@ -29,12 +30,12 @@ import org.apache.pekko.actor.{
Terminated,
Timers
}
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.util.JavaDurationConverters._
-import org.apache.pekko.event.LoggingReceive
-import org.apache.pekko.kafka.KafkaConsumerActor.{ StopLike, StoppingException }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.annotation.InternalApi
+import pekko.util.JavaDurationConverters._
+import pekko.event.LoggingReceive
+import pekko.kafka.KafkaConsumerActor.{ StopLike, StoppingException }
+import pekko.kafka._
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
import org.apache.kafka.clients.consumer._
import org.apache.kafka.common.errors.RebalanceInProgressException
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
@@ -69,7 +70,7 @@ import scala.util.control.NonFatal
final case class RegisterSubStage(tps: Set[TopicPartition]) extends NoSerializationVerificationNeeded
final case class Seek(tps: Map[TopicPartition, Long]) extends NoSerializationVerificationNeeded
final case class RequestMessages(requestId: Int, tps: Set[TopicPartition]) extends NoSerializationVerificationNeeded
- val Stop = org.apache.pekko.kafka.KafkaConsumerActor.Stop
+ val Stop = pekko.kafka.KafkaConsumerActor.Stop
final case class StopFromStage(stageId: String) extends StopLike
final case class Commit(tp: TopicPartition, offsetAndMetadata: OffsetAndMetadata)
extends NoSerializationVerificationNeeded
@@ -407,7 +408,7 @@ import scala.util.control.NonFatal
owner.foreach(_ ! Failure(e))
throw e
case None =>
- import org.apache.pekko.pattern.pipe
+ import pekko.pattern.pipe
implicit val ec: ExecutionContext = context.dispatcher
context.become(expectSettings)
updateSettings.pipeTo(self)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaSourceStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaSourceStage.scala
index f042abb0..e3739d93 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaSourceStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaSourceStage.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.scaladsl.Consumer._
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.stage.{ GraphStageLogic, GraphStageWithMaterializedValue }
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.kafka.scaladsl.Consumer._
+import pekko.stream._
+import pekko.stream.stage.{ GraphStageLogic, GraphStageWithMaterializedValue }
/**
* INTERNAL API
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/LoggingWithId.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/LoggingWithId.scala
index 9d598101..2118fd24 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/LoggingWithId.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/LoggingWithId.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.{ Actor, ActorLogging }
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.stream.stage.{ GraphStageLogic, StageLogging }
+import org.apache.pekko
+import pekko.actor.{ Actor, ActorLogging }
+import pekko.event.LoggingAdapter
+import pekko.stream.stage.{ GraphStageLogic, StageLogging }
/**
* Generate a short random UID for something.
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
index 2bcefd7c..3a39f184 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
@@ -15,10 +15,11 @@
package org.apache.pekko.kafka.internal
import java.util.concurrent.CompletionStage
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ConsumerMessage
-import org.apache.pekko.kafka.ConsumerMessage.{
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.InternalApi
+import pekko.kafka.ConsumerMessage
+import pekko.kafka.ConsumerMessage.{
CommittableMessage,
CommittableOffsetMetadata,
GroupTopicPartition,
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/PartitionAssignmentHelpers.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/PartitionAssignmentHelpers.scala
index 7a7e4970..04f3acc9 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/PartitionAssignmentHelpers.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/PartitionAssignmentHelpers.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
-import org.apache.pekko.kafka.javadsl
-import org.apache.pekko.kafka.{ AutoSubscription, RestrictedConsumer, TopicPartitionsAssigned, TopicPartitionsRevoked }
-import org.apache.pekko.stream.stage.AsyncCallback
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.kafka.javadsl
+import pekko.kafka.{ AutoSubscription, RestrictedConsumer, TopicPartitionsAssigned, TopicPartitionsRevoked }
+import pekko.stream.stage.AsyncCallback
import org.apache.kafka.common.TopicPartition
import scala.jdk.CollectionConverters._
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/PlainSources.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/PlainSources.scala
index ad5d317c..a0a8c14f 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/PlainSources.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/PlainSources.scala
@@ -14,15 +14,16 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.NotUsed
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.{ AutoSubscription, ConsumerSettings, ManualSubscription, Subscription }
-import org.apache.pekko.kafka.internal.SubSourceLogic._
-import org.apache.pekko.stream.SourceShape
-import org.apache.pekko.stream.scaladsl.Source
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.{ AutoSubscription, ConsumerSettings, ManualSubscription, Subscription }
+import pekko.kafka.internal.SubSourceLogic._
+import pekko.stream.SourceShape
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ProducerStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ProducerStage.scala
index ca9bb068..62a1d5a2 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ProducerStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ProducerStage.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.stream._
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.ProducerSettings
+import pekko.stream._
import scala.concurrent.Future
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
index 1e995964..2a51eb20 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.{ ActorRef, ExtendedActorSystem, Terminated }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.Messages
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
-import org.apache.pekko.kafka.{ ConsumerSettings, RestrictedConsumer, Subscription }
-import org.apache.pekko.stream.SourceShape
+import org.apache.pekko
+import pekko.actor.{ ActorRef, ExtendedActorSystem, Terminated }
+import pekko.annotation.InternalApi
+import pekko.kafka.internal.KafkaConsumerActor.Internal.Messages
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.kafka.{ ConsumerSettings, RestrictedConsumer, Subscription }
+import pekko.stream.SourceShape
import org.apache.kafka.common.TopicPartition
import scala.concurrent.{ Future, Promise }
@@ -43,7 +44,7 @@ import scala.concurrent.{ Future, Promise }
final def createConsumerActor(): ActorRef = {
val extendedActorSystem = materializer.system.asInstanceOf[ExtendedActorSystem]
val actor =
- extendedActorSystem.systemActorOf(org.apache.pekko.kafka.KafkaConsumerActor.props(sourceActor.ref, settings),
+ extendedActorSystem.systemActorOf(pekko.kafka.KafkaConsumerActor.props(sourceActor.ref, settings),
s"kafka-consumer-$actorNumber")
consumerPromise.success(actor)
actor
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicBuffer.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicBuffer.scala
index 608caa69..13940e66 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicBuffer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicBuffer.scala
@@ -13,8 +13,9 @@
*/
package org.apache.pekko.kafka.internal
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicSubscription.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicSubscription.scala
index ea1ddc9a..c116a850 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicSubscription.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicSubscription.scala
@@ -13,13 +13,14 @@
*/
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.{ AutoSubscription, ManualSubscription, Subscription }
-import org.apache.pekko.kafka.Subscriptions._
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
-import org.apache.pekko.stream.stage.GraphStageLogic.StageActor
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+import pekko.kafka.{ AutoSubscription, ManualSubscription, Subscription }
+import pekko.kafka.Subscriptions._
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.stream.stage.GraphStageLogic.StageActor
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
import org.apache.kafka.common.TopicPartition
/**
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
index cc54084d..fd9fe846 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
@@ -14,21 +14,22 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.NotUsed
-import org.apache.pekko.actor.Status
-import org.apache.pekko.actor.{ ActorRef, ExtendedActorSystem, Terminated }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.RegisterSubStage
-import org.apache.pekko.kafka.internal.SubSourceLogic._
-import org.apache.pekko.kafka.{ AutoSubscription, ConsumerFailed, ConsumerSettings, RestrictedConsumer }
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
-import org.apache.pekko.pattern.{ ask, AskTimeoutException }
-import org.apache.pekko.stream.scaladsl.Source
-import org.apache.pekko.stream.stage.GraphStageLogic.StageActor
-import org.apache.pekko.stream.stage._
-import org.apache.pekko.stream.{ Attributes, Outlet, SourceShape }
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.Status
+import pekko.actor.{ ActorRef, ExtendedActorSystem, Terminated }
+import pekko.annotation.InternalApi
+import pekko.kafka.internal.KafkaConsumerActor.Internal.RegisterSubStage
+import pekko.kafka.internal.SubSourceLogic._
+import pekko.kafka.{ AutoSubscription, ConsumerFailed, ConsumerSettings, RestrictedConsumer }
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.pattern.{ ask, AskTimeoutException }
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.GraphStageLogic.StageActor
+import pekko.stream.stage._
+import pekko.stream.{ Attributes, Outlet, SourceShape }
+import pekko.util.Timeout
import org.apache.kafka.common.TopicPartition
import scala.annotation.tailrec
@@ -95,7 +96,7 @@ private class SubSourceLogic[K, V, Msg](
}
consumerActor = {
val extendedActorSystem = materializer.system.asInstanceOf[ExtendedActorSystem]
- extendedActorSystem.systemActorOf(org.apache.pekko.kafka.KafkaConsumerActor.props(sourceActor.ref, settings),
+ extendedActorSystem.systemActorOf(pekko.kafka.KafkaConsumerActor.props(sourceActor.ref, settings),
s"kafka-consumer-$actorNumber")
}
consumerPromise.success(consumerActor)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala
index 5411b7ef..cf26a6a1 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala
@@ -14,15 +14,16 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ConsumerMessage.{ GroupTopicPartition, PartitionOffsetCommittedMarker }
-import org.apache.pekko.kafka.ProducerMessage.{ Envelope, Results }
-import org.apache.pekko.kafka.internal.DeferredProducer._
-import org.apache.pekko.kafka.internal.ProducerStage.ProducerCompletionState
-import org.apache.pekko.kafka.{ ConsumerMessage, ProducerSettings }
-import org.apache.pekko.stream.stage._
-import org.apache.pekko.stream.{ Attributes, FlowShape }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.InternalApi
+import pekko.kafka.ConsumerMessage.{ GroupTopicPartition, PartitionOffsetCommittedMarker }
+import pekko.kafka.ProducerMessage.{ Envelope, Results }
+import pekko.kafka.internal.DeferredProducer._
+import pekko.kafka.internal.ProducerStage.ProducerCompletionState
+import pekko.kafka.{ ConsumerMessage, ProducerSettings }
+import pekko.stream.stage._
+import pekko.stream.{ Attributes, FlowShape }
import org.apache.kafka.clients.consumer.{ ConsumerGroupMetadata, OffsetAndMetadata }
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.TopicPartition
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
index ce0fa180..ec6c029f 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
@@ -16,21 +16,22 @@ package org.apache.pekko.kafka.internal
import java.util.Locale
-import org.apache.pekko.{ Done, NotUsed }
-import org.apache.pekko.actor.{ ActorRef, Status, Terminated }
-import org.apache.pekko.actor.Status.Failure
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.Revoked
-import org.apache.pekko.kafka.internal.SubSourceLogic._
-import org.apache.pekko.kafka.internal.TransactionalSubSourceStageLogic.DrainingComplete
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.PartitionAssignmentHandler
-import org.apache.pekko.kafka.{ AutoSubscription, ConsumerFailed, ConsumerSettings, RestrictedConsumer, Subscription }
-import org.apache.pekko.stream.SourceShape
-import org.apache.pekko.stream.scaladsl.Source
-import org.apache.pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.{ Done, NotUsed }
+import pekko.actor.{ ActorRef, Status, Terminated }
+import pekko.actor.Status.Failure
+import pekko.annotation.InternalApi
+import pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
+import pekko.kafka.internal.KafkaConsumerActor.Internal.Revoked
+import pekko.kafka.internal.SubSourceLogic._
+import pekko.kafka.internal.TransactionalSubSourceStageLogic.DrainingComplete
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.PartitionAssignmentHandler
+import pekko.kafka.{ AutoSubscription, ConsumerFailed, ConsumerSettings, RestrictedConsumer, Subscription }
+import pekko.stream.SourceShape
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.{ AsyncCallback, GraphStageLogic }
+import pekko.util.Timeout
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord, OffsetAndMetadata }
import org.apache.kafka.common.{ IsolationLevel, TopicPartition }
@@ -180,7 +181,7 @@ private[internal] abstract class TransactionalSourceLogic[K, V, Msg](shape: Sour
}
private def waitForDraining(partitions: Set[TopicPartition]): Boolean = {
- import org.apache.pekko.pattern.ask
+ import pekko.pattern.ask
implicit val timeout = Timeout(consumerSettings.commitTimeout)
try {
Await.result(ask(stageActor.ref, Drain(partitions, None, Drained)), timeout.duration)
@@ -262,7 +263,7 @@ private[kafka] final class TransactionalSubSource[K, V](
}
private def waitForDraining(partitions: Set[TopicPartition]): Boolean = {
- import org.apache.pekko.pattern.ask
+ import pekko.pattern.ask
implicit val timeout = Timeout(txConsumerSettings.commitTimeout)
try {
val drainCommandFutures =
@@ -294,7 +295,7 @@ private object TransactionalSourceLogic {
private[internal] final case class CommittedMarkerRef(sourceActor: ActorRef, commitTimeout: FiniteDuration)(
implicit ec: ExecutionContext) extends CommittedMarker {
override def committed(offsets: Map[TopicPartition, OffsetAndMetadata]): Future[Done] = {
- import org.apache.pekko.pattern.ask
+ import pekko.pattern.ask
sourceActor
.ask(Committed(offsets))(Timeout(commitTimeout))
.map(_ => Done)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
index f0f8c101..99febc06 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
@@ -15,12 +15,13 @@
package org.apache.pekko.kafka.javadsl
import java.util.concurrent.CompletionStage
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.japi.Pair
-import org.apache.pekko.{ Done, NotUsed }
-import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
-import org.apache.pekko.kafka.{ scaladsl, CommitterSettings }
-import org.apache.pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.japi.Pair
+import pekko.{ Done, NotUsed }
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
+import pekko.kafka.{ scaladsl, CommitterSettings }
+import pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
import scala.compat.java8.FutureConverters.FutureOps
@@ -63,10 +64,10 @@ object Committer {
@ApiMayChange
def sinkWithOffsetContext[E, C <: Committable](
settings: CommitterSettings): Sink[Pair[E, C], CompletionStage[Done]] =
- org.apache.pekko.stream.scaladsl
+ pekko.stream.scaladsl
.Flow[Pair[E, C]]
.map(_.toScala)
- .toMat(scaladsl.Committer.sinkWithOffsetContext(settings))(org.apache.pekko.stream.scaladsl.Keep.right)
+ .toMat(scaladsl.Committer.sinkWithOffsetContext(settings))(pekko.stream.scaladsl.Keep.right)
.mapMaterializedValue[CompletionStage[Done]](_.toJava)
.asJava[Pair[E, C]]
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
index 821e4ac3..3e4323cc 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
@@ -16,15 +16,16 @@ package org.apache.pekko.kafka.javadsl
import java.util.concurrent.{ CompletionStage, Executor }
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.japi.Pair
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.internal.{ ConsumerControlAsJava, SourceWithOffsetContext }
-import org.apache.pekko.stream.javadsl.{ Source, SourceWithContext }
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.ApiMayChange
+import pekko.dispatch.ExecutionContexts
+import pekko.japi.Pair
+import pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
+import pekko.kafka._
+import pekko.kafka.internal.{ ConsumerControlAsJava, SourceWithOffsetContext }
+import pekko.stream.javadsl.{ Source, SourceWithContext }
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
@@ -186,7 +187,7 @@ object Consumer {
subscription: Subscription): SourceWithContext[ConsumerRecord[K, V], CommittableOffset, Control] =
// TODO this could use `scaladsl committableSourceWithContext` but `mapMaterializedValue` is not available, yet
// See https://github.com/akka/akka/issues/26836
- org.apache.pekko.stream.scaladsl.Source
+ pekko.stream.scaladsl.Source
.fromGraph(new SourceWithOffsetContext[K, V](settings, subscription))
.mapMaterializedValue(ConsumerControlAsJava.apply)
.asSourceWithContext(_._2)
@@ -216,7 +217,7 @@ object Consumer {
: SourceWithContext[ConsumerRecord[K, V], CommittableOffset, Control] =
// TODO this could use `scaladsl committableSourceWithContext` but `mapMaterializedValue` is not available, yet
// See https://github.com/akka/akka/issues/26836
- org.apache.pekko.stream.scaladsl.Source
+ pekko.stream.scaladsl.Source
.fromGraph(
new SourceWithOffsetContext[K, V](settings,
subscription,
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
index 6b2a04fa..11883b87 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
@@ -16,8 +16,9 @@ package org.apache.pekko.kafka.javadsl
import java.util.concurrent.CompletionStage
-import org.apache.pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
-import org.apache.pekko.kafka.{ scaladsl, ConsumerSettings, ProducerSettings }
+import org.apache.pekko
+import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
+import pekko.kafka.{ scaladsl, ConsumerSettings, ProducerSettings }
import com.typesafe.config.Config
import scala.compat.java8.FunctionConverters._
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
index fe767c57..2bcb2e12 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
@@ -16,10 +16,11 @@ package org.apache.pekko.kafka.javadsl
import java.util.concurrent.{ CompletionStage, Executor }
-import org.apache.pekko.actor.{ ActorRef, ActorSystem }
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.ConsumerSettings
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.actor.{ ActorRef, ActorSystem }
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.ConsumerSettings
+import pekko.util.Timeout
import org.apache.kafka.clients.consumer.OffsetAndMetadata
import org.apache.kafka.common.{ PartitionInfo, TopicPartition }
@@ -27,7 +28,7 @@ import scala.compat.java8.FutureConverters._
import scala.concurrent.ExecutionContextExecutor
import scala.jdk.CollectionConverters._
-class MetadataClient private (metadataClient: org.apache.pekko.kafka.scaladsl.MetadataClient) {
+class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClient) {
def getBeginningOffsets[K, V](
partitions: java.util.Set[TopicPartition]): CompletionStage[java.util.Map[TopicPartition, java.lang.Long]] =
@@ -98,7 +99,7 @@ object MetadataClient {
def create(consumerActor: ActorRef, timeout: Timeout, executor: Executor): MetadataClient = {
implicit val ec: ExecutionContextExecutor = ExecutionContexts.fromExecutor(executor)
- val metadataClient = org.apache.pekko.kafka.scaladsl.MetadataClient.create(consumerActor, timeout)
+ val metadataClient = pekko.kafka.scaladsl.MetadataClient.create(consumerActor, timeout)
new MetadataClient(metadataClient)
}
@@ -106,7 +107,7 @@ object MetadataClient {
timeout: Timeout,
system: ActorSystem,
executor: Executor): MetadataClient = {
- val metadataClient = org.apache.pekko.kafka.scaladsl.MetadataClient
+ val metadataClient = pekko.kafka.scaladsl.MetadataClient
.create(consumerSettings, timeout)(system, ExecutionContexts.fromExecutor(executor))
new MetadataClient(metadataClient)
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/PartitionAssignmentHandler.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/PartitionAssignmentHandler.scala
index 9cd2a2ac..2ca12d4c 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/PartitionAssignmentHandler.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/PartitionAssignmentHandler.scala
@@ -14,7 +14,8 @@
package org.apache.pekko.kafka.javadsl
-import org.apache.pekko.kafka.RestrictedConsumer
+import org.apache.pekko
+import pekko.kafka.RestrictedConsumer
import org.apache.kafka.common.TopicPartition
/**
@@ -36,7 +37,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsRevoked]]
*
* @param revokedTps The list of partitions that were revoked from the consumer
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onRevoke(revokedTps: java.util.Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -44,7 +45,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsAssigned]]
*
* @param assignedTps The list of partitions that are now assigned to the consumer (may include partitions previously assigned to the consumer)
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onAssign(assignedTps: java.util.Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -53,7 +54,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsLost]]
*
* @param lostTps The list of partitions that are no longer valid
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onLost(lostTps: java.util.Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -62,7 +63,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsRevoked]]
*
* @param currentTps The list of partitions that are currently assigned to the consumer
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onStop(currentTps: java.util.Set[TopicPartition], consumer: RestrictedConsumer): Unit
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
index 3a5a4e79..4aca27c0 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
@@ -15,12 +15,13 @@
package org.apache.pekko.kafka.javadsl
import java.util.concurrent.CompletionStage
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.ConsumerMessage.Committable
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.{ scaladsl, CommitterSettings, ConsumerMessage, ProducerSettings }
-import org.apache.pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
-import org.apache.pekko.{ japi, Done, NotUsed }
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.kafka.ConsumerMessage.Committable
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.{ scaladsl, CommitterSettings, ConsumerMessage, ProducerSettings }
+import pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
+import pekko.{ japi, Done, NotUsed }
import org.apache.kafka.clients.producer.ProducerRecord
import scala.annotation.nowarn
@@ -66,11 +67,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -95,11 +96,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is always a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -121,11 +122,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -144,11 +145,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -156,16 +157,16 @@ object Producer {
@ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/880")
def committableSinkWithOffsetContext[K, V, IN <: Envelope[K, V, _], C <: Committable](
producerSettings: ProducerSettings[K, V],
- committerSettings: CommitterSettings): Sink[org.apache.pekko.japi.Pair[IN, C], CompletionStage[Done]] =
+ committerSettings: CommitterSettings): Sink[pekko.japi.Pair[IN, C], CompletionStage[Done]] =
committableSink(producerSettings, committerSettings)
- .contramap(new org.apache.pekko.japi.function.Function[japi.Pair[IN, C], Envelope[K, V, C]] {
+ .contramap(new pekko.japi.function.Function[japi.Pair[IN, C], Envelope[K, V, C]] {
override def apply(p: japi.Pair[IN, C]) = p.first.withPassThrough(p.second)
})
/**
* Create a flow to publish records to Kafka topics and then pass it on.
*
- * The records must be wrapped in a [[org.apache.pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]].
+ * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -189,11 +190,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -213,11 +214,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
*
@@ -231,7 +232,7 @@ object Producer {
/**
* Create a flow to publish records to Kafka topics and then pass it on.
*
- * The records must be wrapped in a [[org.apache.pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]].
+ * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -253,11 +254,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -281,11 +282,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
*
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
index 3694a3dc..774603f0 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
@@ -16,10 +16,11 @@ package org.apache.pekko.kafka.javadsl
import java.util.concurrent.CompletionStage
-import org.apache.pekko.Done
-import org.apache.pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.{ scaladsl, ProducerSettings }
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.{ scaladsl, ProducerSettings }
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
import scala.compat.java8.FutureConverters._
@@ -48,11 +49,11 @@ final class SendProducer[K, V] private (underlying: scaladsl.SendProducer[K, V])
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and completes the future with [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and completes the future with [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and completes the future with [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support passing through arbitrary data.
*/
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
index cbdd66d9..fb6fe2c8 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
@@ -16,15 +16,16 @@ package org.apache.pekko.kafka.javadsl
import java.util.concurrent.CompletionStage
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.japi.Pair
-import org.apache.pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.internal.{ ConsumerControlAsJava, TransactionalSourceWithOffsetContext }
-import org.apache.pekko.kafka.javadsl.Consumer.Control
-import org.apache.pekko.stream.javadsl._
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.japi.Pair
+import pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka._
+import pekko.kafka.internal.{ ConsumerControlAsJava, TransactionalSourceWithOffsetContext }
+import pekko.kafka.javadsl.Consumer.Control
+import pekko.stream.javadsl._
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerRecord
import scala.compat.java8.FutureConverters.FutureOps
@@ -55,7 +56,7 @@ object Transactional {
def sourceWithOffsetContext[K, V](
consumerSettings: ConsumerSettings[K, V],
subscription: Subscription): SourceWithContext[ConsumerRecord[K, V], PartitionOffset, Control] =
- org.apache.pekko.stream.scaladsl.Source
+ pekko.stream.scaladsl.Source
.fromGraph(new TransactionalSourceWithOffsetContext[K, V](consumerSettings, subscription))
.mapMaterializedValue(ConsumerControlAsJava.apply)
.asSourceWithContext(_._2)
@@ -109,11 +110,11 @@ object Transactional {
def sinkWithOffsetContext[K, V](
settings: ProducerSettings[K, V],
transactionalId: String): Sink[Pair[Envelope[K, V, NotUsed], PartitionOffset], CompletionStage[Done]] =
- org.apache.pekko.stream.scaladsl
+ pekko.stream.scaladsl
.Flow[Pair[Envelope[K, V, NotUsed], PartitionOffset]]
.map(_.toScala)
.toMat(scaladsl.Transactional.sinkWithOffsetContext(settings, transactionalId))(
- org.apache.pekko.stream.scaladsl.Keep.right)
+ pekko.stream.scaladsl.Keep.right)
.mapMaterializedValue(_.toJava)
.asJava
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Committer.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Committer.scala
index 3f56aecd..ca9462d6 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Committer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Committer.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.CommitterSettings
-import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
-import org.apache.pekko.kafka.internal.CommitCollectorStage
-import org.apache.pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink }
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.CommitterSettings
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
+import pekko.kafka.internal.CommitCollectorStage
+import pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink }
+import pekko.{ Done, NotUsed }
import scala.concurrent.Future
@@ -41,7 +42,7 @@ object Committer {
.fromGraph(new CommitCollectorStage(settings))
// See https://github.com/akka/alpakka-kafka/issues/882
- import org.apache.pekko.kafka.CommitDelivery._
+ import pekko.kafka.CommitDelivery._
settings.delivery match {
case WaitForAck =>
offsetBatches
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Consumer.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Consumer.scala
index 1a760a4e..576b94d2 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Consumer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Consumer.scala
@@ -14,14 +14,15 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.internal._
-import org.apache.pekko.stream.scaladsl.{ Source, SourceWithContext }
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.ApiMayChange
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
+import pekko.kafka._
+import pekko.kafka.internal._
+import pekko.stream.scaladsl.{ Source, SourceWithContext }
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/DiscoverySupport.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/DiscoverySupport.scala
index 0b41286d..698f70e3 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/DiscoverySupport.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/DiscoverySupport.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.actor.{ ActorSystem, ActorSystemImpl, ClassicActorSystemProvider }
-import org.apache.pekko.annotation.InternalApi
-import org.apache.pekko.discovery.{ Discovery, ServiceDiscovery }
-import org.apache.pekko.kafka.{ ConsumerSettings, ProducerSettings }
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.actor.{ ActorSystem, ActorSystemImpl, ClassicActorSystemProvider }
+import pekko.annotation.InternalApi
+import pekko.discovery.{ Discovery, ServiceDiscovery }
+import pekko.kafka.{ ConsumerSettings, ProducerSettings }
+import pekko.util.JavaDurationConverters._
import com.typesafe.config.Config
import scala.concurrent.Future
@@ -28,7 +29,7 @@ import scala.util.Failure
/**
* Scala API.
*
- * Reads Kafka bootstrap servers from configured sources via [[org.apache.pekko.discovery.Discovery]] configuration.
+ * Reads Kafka bootstrap servers from configured sources via [[pekko.discovery.Discovery]] configuration.
*/
object DiscoverySupport {
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
index b171b907..997ce48b 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
@@ -16,12 +16,13 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicLong
-import org.apache.pekko.actor.{ ActorRef, ActorSystem, ExtendedActorSystem }
-import org.apache.pekko.dispatch.ExecutionContexts
-import org.apache.pekko.kafka.Metadata._
-import org.apache.pekko.kafka.{ ConsumerSettings, KafkaConsumerActor }
-import org.apache.pekko.pattern.ask
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.actor.{ ActorRef, ActorSystem, ExtendedActorSystem }
+import pekko.dispatch.ExecutionContexts
+import pekko.kafka.Metadata._
+import pekko.kafka.{ ConsumerSettings, KafkaConsumerActor }
+import pekko.pattern.ask
+import pekko.util.Timeout
import org.apache.kafka.clients.consumer.OffsetAndMetadata
import org.apache.kafka.common.{ PartitionInfo, TopicPartition }
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/PartitionAssignmentHandler.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/PartitionAssignmentHandler.scala
index 8129af2b..478b0f9b 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/PartitionAssignmentHandler.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/PartitionAssignmentHandler.scala
@@ -14,7 +14,8 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.kafka.RestrictedConsumer
+import org.apache.pekko
+import pekko.kafka.RestrictedConsumer
import org.apache.kafka.common.TopicPartition
/**
@@ -25,7 +26,7 @@ import org.apache.kafka.common.TopicPartition
* A warning will be logged if a callback takes longer than the configured `partition-handler-warning`.
*
* There is no point in calling `Committable`'s commit methods as their committing won't be executed as long as any of
- * the callbacks in this class are called. Calling `commitSync` on the passed [[org.apache.pekko.kafka.RestrictedConsumer]] is available.
+ * the callbacks in this class are called. Calling `commitSync` on the passed [[pekko.kafka.RestrictedConsumer]] is available.
*
* This complements the methods of Kafka's [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener ConsumerRebalanceListener]] with
* an `onStop` callback which is called before `Consumer.close`.
@@ -36,7 +37,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsRevoked]]
*
* @param revokedTps The list of partitions that were revoked from the consumer
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onRevoke(revokedTps: Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -44,7 +45,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsAssigned]]
*
* @param assignedTps The list of partitions that are now assigned to the consumer (may include partitions previously assigned to the consumer)
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onAssign(assignedTps: Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -53,7 +54,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsLost]]
*
* @param lostTps The list of partitions that are no longer valid
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onLost(lostTps: Set[TopicPartition], consumer: RestrictedConsumer): Unit
@@ -62,7 +63,7 @@ trait PartitionAssignmentHandler {
* See [[org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsRevoked]]
*
* @param currentTps The list of partitions that are currently assigned to the consumer
- * @param consumer The [[org.apache.pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
+ * @param consumer The [[pekko.kafka.RestrictedConsumer]] gives some access to the internally used [[org.apache.kafka.clients.consumer.Consumer Consumer]]
*/
def onStop(currentTps: Set[TopicPartition], consumer: RestrictedConsumer): Unit
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
index e6b8004f..e25ea423 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
@@ -14,14 +14,15 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.ConsumerMessage.Committable
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.internal.{ CommittingProducerSinkStage, DefaultProducerStage }
-import org.apache.pekko.kafka.{ CommitterSettings, ConsumerMessage, ProducerSettings }
-import org.apache.pekko.stream.ActorAttributes
-import org.apache.pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink }
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.kafka.ConsumerMessage.Committable
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.internal.{ CommittingProducerSinkStage, DefaultProducerStage }
+import pekko.kafka.{ CommitterSettings, ConsumerMessage, ProducerSettings }
+import pekko.stream.ActorAttributes
+import pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink }
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.producer.ProducerRecord
import scala.concurrent.Future
@@ -66,11 +67,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -89,11 +90,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is always a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -113,11 +114,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -133,11 +134,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
*
* Note that there is a risk that something fails after publishing but before
* committing, so it is "at-least once delivery" semantics.
@@ -155,7 +156,7 @@ object Producer {
/**
* Create a flow to publish records to Kafka topics and then pass it on.
*
- * The records must be wrapped in a [[org.apache.pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]].
+ * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -178,11 +179,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -206,11 +207,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
*
@@ -227,7 +228,7 @@ object Producer {
/**
* Create a flow to publish records to Kafka topics and then pass it on.
*
- * The records must be wrapped in a [[org.apache.pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]].
+ * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -247,11 +248,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
* or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
@@ -275,11 +276,11 @@ object Producer {
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
*
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
index ddb31571..0a5ae02e 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.ProducerSettings
+import pekko.util.JavaDurationConverters._
import org.apache.kafka.clients.producer.{ Callback, ProducerRecord, RecordMetadata }
import scala.concurrent.{ ExecutionContext, Future, Promise }
@@ -37,11 +38,11 @@ final class SendProducer[K, V] private (val settings: ProducerSettings[K, V], sy
*
* It publishes records to Kafka topics conditionally:
*
- * - [[org.apache.pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.Result Result]]
+ * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and completes the future with [[pekko.kafka.ProducerMessage.Result Result]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.MultiResult MultiResult]]
+ * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and completes the future with [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
*
- * - [[org.apache.pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and completes the future with [[org.apache.pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
+ * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and completes the future with [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
*
* The messages support passing through arbitrary data.
*/
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Transactional.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Transactional.scala
index f77bc90e..2af18f9a 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Transactional.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Transactional.scala
@@ -14,20 +14,21 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.annotation.{ ApiMayChange, InternalApi }
-import org.apache.pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.internal.{
+import org.apache.pekko
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.internal.{
TransactionalProducerStage,
TransactionalSource,
TransactionalSourceWithOffsetContext,
TransactionalSubSource
}
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.{ AutoSubscription, ConsumerMessage, ConsumerSettings, ProducerSettings, Subscription }
-import org.apache.pekko.stream.ActorAttributes
-import org.apache.pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink, Source, SourceWithContext }
-import org.apache.pekko.{ Done, NotUsed }
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.{ AutoSubscription, ConsumerMessage, ConsumerSettings, ProducerSettings, Subscription }
+import pekko.stream.ActorAttributes
+import pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink, Source, SourceWithContext }
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ConsumerResultFactory.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ConsumerResultFactory.scala
index bb8ded6e..cde15eed 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ConsumerResultFactory.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ConsumerResultFactory.scala
@@ -14,18 +14,19 @@
package org.apache.pekko.kafka.testkit
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.ConsumerMessage
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableOffset, GroupTopicPartition, PartitionOffsetCommittedMarker }
-import org.apache.pekko.kafka.internal.{ CommittableOffsetImpl, KafkaAsyncConsumerCommitterRef }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.ApiMayChange
+import pekko.kafka.ConsumerMessage
+import pekko.kafka.ConsumerMessage.{ CommittableOffset, GroupTopicPartition, PartitionOffsetCommittedMarker }
+import pekko.kafka.internal.{ CommittableOffsetImpl, KafkaAsyncConsumerCommitterRef }
import org.apache.kafka.clients.consumer.{ ConsumerRecord, OffsetAndMetadata }
import org.apache.kafka.common.TopicPartition
import scala.concurrent.Future
/**
- * Factory methods to create instances that normally are emitted by [[org.apache.pekko.kafka.scaladsl.Consumer]] and [[org.apache.pekko.kafka.javadsl.Consumer]] flows.
+ * Factory methods to create instances that normally are emitted by [[pekko.kafka.scaladsl.Consumer]] and [[pekko.kafka.javadsl.Consumer]] flows.
*/
@ApiMayChange
object ConsumerResultFactory {
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala
index 8ccc4627..f7e50e17 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala
@@ -16,10 +16,11 @@ package org.apache.pekko.kafka.testkit
import java.time.Duration
import java.util.function.Consumer
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.util.JavaDurationConverters._
import com.typesafe.config.Config
-import org.apache.pekko.kafka.testkit.internal.PekkoConnectorsKafkaContainer
+import pekko.kafka.testkit.internal.PekkoConnectorsKafkaContainer
import org.testcontainers.containers.GenericContainer
import scala.concurrent.duration.FiniteDuration
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ProducerResultFactory.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ProducerResultFactory.scala
index e4ebe012..77e8103b 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ProducerResultFactory.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/ProducerResultFactory.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka.testkit
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.ProducerMessage
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.kafka.ProducerMessage
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
import org.apache.kafka.common.TopicPartition
@@ -23,7 +24,7 @@ import scala.jdk.CollectionConverters._
import scala.collection.immutable
/**
- * Factory methods to create instances that normally are emitted by [[org.apache.pekko.kafka.scaladsl.Producer]] and [[org.apache.pekko.kafka.javadsl.Producer]] flows.
+ * Factory methods to create instances that normally are emitted by [[pekko.kafka.scaladsl.Producer]] and [[pekko.kafka.javadsl.Producer]] flows.
*/
@ApiMayChange
object ProducerResultFactory {
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
index 6175e13a..fbe74858 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
@@ -19,9 +19,10 @@ import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.Arrays
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.testkit.KafkaTestkitSettings
-import org.apache.pekko.kafka.{ CommitterSettings, ConsumerSettings, ProducerSettings }
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.testkit.KafkaTestkitSettings
+import pekko.kafka.{ CommitterSettings, ConsumerSettings, ProducerSettings }
import org.apache.kafka.clients.admin.{ Admin, AdminClientConfig, NewTopic }
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.{ Deserializer, Serializer, StringDeserializer, StringSerializer }
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
index b843e23a..ac4f1208 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka.testkit.internal
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.KafkaSpec
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.KafkaSpec
+import pekko.util.JavaDurationConverters._
import org.testcontainers.containers.GenericContainer
import org.testcontainers.utility.DockerImageName
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/javadsl/ConsumerControlFactory.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/javadsl/ConsumerControlFactory.scala
index 69d8c8bf..63445997 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/javadsl/ConsumerControlFactory.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/javadsl/ConsumerControlFactory.scala
@@ -16,15 +16,16 @@ package org.apache.pekko.kafka.testkit.javadsl
import java.util.concurrent.{ CompletableFuture, CompletionStage, Executor }
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.javadsl.Consumer
-import org.apache.pekko.stream.javadsl.{ Flow, Keep, Source }
-import org.apache.pekko.stream.{ scaladsl, KillSwitch, KillSwitches }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.ApiMayChange
+import pekko.kafka.javadsl.Consumer
+import pekko.stream.javadsl.{ Flow, Keep, Source }
+import pekko.stream.{ scaladsl, KillSwitch, KillSwitches }
import org.apache.kafka.common.{ Metric, MetricName }
/**
- * Helper factory to create [[org.apache.pekko.kafka.javadsl.Consumer.Control]] instances when
+ * Helper factory to create [[pekko.kafka.javadsl.Consumer.Control]] instances when
* testing without a Kafka broker.
*/
@ApiMayChange
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/ConsumerControlFactory.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/ConsumerControlFactory.scala
index cf9ae496..7dece911 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/ConsumerControlFactory.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/ConsumerControlFactory.scala
@@ -14,17 +14,18 @@
package org.apache.pekko.kafka.testkit.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.annotation.ApiMayChange
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.stream.scaladsl.{ Flow, Keep, Source }
-import org.apache.pekko.stream.{ KillSwitch, KillSwitches }
+import org.apache.pekko
+import pekko.Done
+import pekko.annotation.ApiMayChange
+import pekko.kafka.scaladsl.Consumer
+import pekko.stream.scaladsl.{ Flow, Keep, Source }
+import pekko.stream.{ KillSwitch, KillSwitches }
import org.apache.kafka.common.{ Metric, MetricName }
import scala.concurrent.{ Future, Promise }
/**
- * Helper factory to create [[org.apache.pekko.kafka.scaladsl.Consumer.Control]] instances when
+ * Helper factory to create [[pekko.kafka.scaladsl.Consumer.Control]] instances when
* testing without a Kafka broker.
*/
@ApiMayChange
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
index 88537f82..638a9966 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
@@ -18,18 +18,19 @@ import java.time.Duration
import java.util
import java.util.concurrent.TimeUnit
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer }
-import org.apache.pekko.kafka.testkit.internal.{ KafkaTestKit, KafkaTestKitChecks }
-import org.apache.pekko.stream.{ Materializer, SystemMaterializer }
-import org.apache.pekko.stream.scaladsl.{ Keep, Source }
-import org.apache.pekko.stream.testkit.TestSubscriber
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.event.LoggingAdapter
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.{ Consumer, Producer }
+import pekko.kafka.testkit.internal.{ KafkaTestKit, KafkaTestKitChecks }
+import pekko.stream.{ Materializer, SystemMaterializer }
+import pekko.stream.scaladsl.{ Keep, Source }
+import pekko.stream.testkit.TestSubscriber
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
import org.apache.kafka.clients.admin._
import org.apache.kafka.clients.producer.{ Producer => KProducer, ProducerRecord }
import org.apache.kafka.common.ConsumerGroupState
@@ -56,7 +57,7 @@ abstract class KafkaSpec(_kafkaPort: Int, val zooKeeperPort: Int, actorSystem: A
implicit val ec: ExecutionContext = system.dispatcher
implicit val mat: Materializer = SystemMaterializer(system).materializer
- implicit val scheduler: org.apache.pekko.actor.Scheduler = system.scheduler
+ implicit val scheduler: pekko.actor.Scheduler = system.scheduler
var testProducer: KProducer[String, String] = _
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/TestcontainersKafkaLike.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/TestcontainersKafkaLike.scala
index 26fb0def..aa4a29c1 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/TestcontainersKafkaLike.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/TestcontainersKafkaLike.scala
@@ -14,12 +14,9 @@
package org.apache.pekko.kafka.testkit.scaladsl
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.internal.{
- PekkoConnectorsKafkaContainer,
- SchemaRegistryContainer,
- TestcontainersKafka
-}
+import org.apache.pekko
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.internal.{ PekkoConnectorsKafkaContainer, SchemaRegistryContainer, TestcontainersKafka }
import org.testcontainers.containers.GenericContainer
/**
diff --git a/tests/src/it/scala/org/apache/pekko/kafka/IntegrationTests.scala b/tests/src/it/scala/org/apache/pekko/kafka/IntegrationTests.scala
index 63737b6a..bd64c148 100644
--- a/tests/src/it/scala/org/apache/pekko/kafka/IntegrationTests.scala
+++ b/tests/src/it/scala/org/apache/pekko/kafka/IntegrationTests.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka
-import org.apache.pekko.NotUsed
-import org.apache.pekko.stream.scaladsl.Flow
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.stream.scaladsl.Flow
import org.apache.kafka.common.TopicPartition
import org.slf4j.Logger
import org.testcontainers.containers.GenericContainer
diff --git a/tests/src/it/scala/org/apache/pekko/kafka/PartitionedSourceFailoverSpec.scala b/tests/src/it/scala/org/apache/pekko/kafka/PartitionedSourceFailoverSpec.scala
index c026972e..75010572 100644
--- a/tests/src/it/scala/org/apache/pekko/kafka/PartitionedSourceFailoverSpec.scala
+++ b/tests/src/it/scala/org/apache/pekko/kafka/PartitionedSourceFailoverSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.scaladsl.{ Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{ ProducerConfig, ProducerRecord }
import org.apache.kafka.common.config.TopicConfig
diff --git a/tests/src/it/scala/org/apache/pekko/kafka/PlainSourceFailoverSpec.scala b/tests/src/it/scala/org/apache/pekko/kafka/PlainSourceFailoverSpec.scala
index 349be3fe..462db4e8 100644
--- a/tests/src/it/scala/org/apache/pekko/kafka/PlainSourceFailoverSpec.scala
+++ b/tests/src/it/scala/org/apache/pekko/kafka/PlainSourceFailoverSpec.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.scaladsl.{ Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{ ProducerConfig, ProducerRecord }
import org.apache.kafka.common.config.TopicConfig
diff --git a/tests/src/it/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala b/tests/src/it/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
index 49bf7e49..b09ed446 100644
--- a/tests/src/it/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
+++ b/tests/src/it/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
@@ -16,13 +16,14 @@ package org.apache.pekko.kafka
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.scaladsl.SpecBase
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.scaladsl.SpecBase
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream._
+import pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.scalatest.concurrent.PatienceConfiguration.Interval
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.Ignore
diff --git a/tests/src/it/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala b/tests/src/it/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
index cabc36d2..a9e76eed 100644
--- a/tests/src/it/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
+++ b/tests/src/it/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
@@ -16,14 +16,15 @@ package org.apache.pekko.kafka
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.{ Consumer, SpecBase, Transactional }
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.scaladsl.{ Flow, Keep, RestartSource, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.{ Consumer, SpecBase, Transactional }
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream._
+import pekko.stream.scaladsl.{ Flow, Keep, RestartSource, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import org.scalatest.concurrent.PatienceConfiguration.Interval
import org.scalatest.concurrent.ScalaFutures
diff --git a/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala b/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
index 51404c9c..bf4047d2 100644
--- a/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
@@ -14,12 +14,13 @@
package docs.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.Subscriptions
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.scaladsl.{ Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.Subscriptions
+import pekko.kafka.scaladsl.{ Consumer, Producer, SpecBase }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
diff --git a/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala b/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
index 6e6d57f9..24d62597 100644
--- a/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
+++ b/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
@@ -15,15 +15,16 @@
package docs.scaladsl
// #oneToMany
-import org.apache.pekko.{ Done, NotUsed }
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableOffset, CommittableOffsetBatch }
-import org.apache.pekko.kafka.ProducerMessage.Envelope
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.{ ProducerMessage, Subscriptions }
-import org.apache.pekko.kafka.scaladsl.{ Committer, Consumer, Producer }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.{ Done, NotUsed }
+import pekko.kafka.ConsumerMessage.{ CommittableOffset, CommittableOffsetBatch }
+import pekko.kafka.ProducerMessage.Envelope
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.{ ProducerMessage, Subscriptions }
+import pekko.kafka.scaladsl.{ Committer, Consumer, Producer }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Keep, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import scala.collection.immutable
diff --git a/tests/src/test/scala/docs/scaladsl/ClusterShardingExample.scala b/tests/src/test/scala/docs/scaladsl/ClusterShardingExample.scala
index b7d48998..c12ad5cd 100644
--- a/tests/src/test/scala/docs/scaladsl/ClusterShardingExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/ClusterShardingExample.scala
@@ -14,17 +14,18 @@
package docs.scaladsl
-import org.apache.pekko.NotUsed
-import org.apache.pekko.actor.typed.scaladsl.Behaviors
-import org.apache.pekko.actor.typed.scaladsl.adapter._
-import org.apache.pekko.actor.typed.{ ActorSystem, Behavior }
-import org.apache.pekko.cluster.sharding.external.ExternalShardAllocationStrategy
-import org.apache.pekko.cluster.sharding.typed.ClusterShardingSettings
-import org.apache.pekko.cluster.sharding.typed.scaladsl.{ ClusterSharding, Entity, EntityTypeKey }
-import org.apache.pekko.kafka.cluster.sharding.KafkaClusterSharding
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.kafka.{ ConsumerRebalanceEvent, ConsumerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl.{ Flow, Sink }
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.actor.typed.{ ActorSystem, Behavior }
+import pekko.cluster.sharding.external.ExternalShardAllocationStrategy
+import pekko.cluster.sharding.typed.ClusterShardingSettings
+import pekko.cluster.sharding.typed.scaladsl.{ ClusterSharding, Entity, EntityTypeKey }
+import pekko.kafka.cluster.sharding.KafkaClusterSharding
+import pekko.kafka.scaladsl.Consumer
+import pekko.kafka.{ ConsumerRebalanceEvent, ConsumerSettings, Subscriptions }
+import pekko.stream.scaladsl.{ Flow, Sink }
import org.apache.kafka.common.serialization.{ ByteArrayDeserializer, StringDeserializer }
import scala.concurrent.Future
@@ -79,12 +80,12 @@ object ClusterShardingExample {
// #rebalance-listener
// obtain an Apache Pekko classic ActorRef that will handle consumer group rebalance events
- val rebalanceListener: org.apache.pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] =
+ val rebalanceListener: pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] =
KafkaClusterSharding(system.toClassic).rebalanceListener(typeKey)
// convert the rebalance listener to a classic ActorRef until Apache Pekko Connector Kafka supports Apache Pekko Typed
- import org.apache.pekko.actor.typed.scaladsl.adapter._
- val rebalanceListenerClassic: org.apache.pekko.actor.ActorRef = rebalanceListener.toClassic
+ import pekko.actor.typed.scaladsl.adapter._
+ val rebalanceListenerClassic: pekko.actor.ActorRef = rebalanceListener.toClassic
val consumerSettings =
ConsumerSettings(system.toClassic, new StringDeserializer, new ByteArrayDeserializer)
diff --git a/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala b/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
index 010f6328..9dcf82bb 100644
--- a/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
@@ -16,17 +16,18 @@ package docs.scaladsl
import java.util.concurrent.atomic.{ AtomicLong, AtomicReference }
-import org.apache.pekko.Done
-import org.apache.pekko.actor.typed.Behavior
-import org.apache.pekko.actor.typed.scaladsl.Behaviors
-import org.apache.pekko.actor.{ Actor, ActorLogging, Props }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.scaladsl._
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.RestartSettings
-import org.apache.pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.{ Actor, ActorLogging, Props }
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.scaladsl._
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.RestartSettings
+import pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord }
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
@@ -345,7 +346,8 @@ class ConsumerExample extends DocsSpecBase with TestcontainersKafkaLike {
val revokedPromise = Promise[Done]()
// format: off
//#withRebalanceListenerActor
- import org.apache.pekko.kafka.{TopicPartitionsAssigned, TopicPartitionsRevoked}
+ import org.apache.pekko
+ import pekko.kafka.{TopicPartitionsAssigned, TopicPartitionsRevoked}
class RebalanceListener extends Actor with ActorLogging {
def receive: Receive = {
@@ -394,7 +396,8 @@ class ConsumerExample extends DocsSpecBase with TestcontainersKafkaLike {
// format: off
//#withTypedRebalanceListenerActor
- import org.apache.pekko.kafka.{TopicPartitionsAssigned, TopicPartitionsRevoked}
+ import org.apache.pekko
+ import pekko.kafka.{TopicPartitionsAssigned, TopicPartitionsRevoked}
def rebalanceListener(): Behavior[ConsumerRebalanceEvent] = Behaviors.receive {
case (context, TopicPartitionsAssigned(subscription, topicPartitions)) =>
@@ -416,13 +419,13 @@ class ConsumerExample extends DocsSpecBase with TestcontainersKafkaLike {
val guardian = Behaviors.setup[Nothing] { context =>
//#withTypedRebalanceListenerActor
- val typedRef: org.apache.pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] =
+ val typedRef: pekko.actor.typed.ActorRef[ConsumerRebalanceEvent] =
context.spawn(rebalanceListener(), "rebalance-listener")
// adds support for actors to a classic actor system and context
- import org.apache.pekko.actor.typed.scaladsl.adapter._
+ import pekko.actor.typed.scaladsl.adapter._
- val classicRef: org.apache.pekko.actor.ActorRef = typedRef.toClassic
+ val classicRef: pekko.actor.ActorRef = typedRef.toClassic
val subscription = Subscriptions
.topics(topic)
@@ -449,7 +452,7 @@ class ConsumerExample extends DocsSpecBase with TestcontainersKafkaLike {
}
// fixme: get typed system from existing `system`
- val typed = org.apache.pekko.actor.typed.ActorSystem[Nothing](guardian, "typed-rebalance-listener-example")
+ val typed = pekko.actor.typed.ActorSystem[Nothing](guardian, "typed-rebalance-listener-example")
typed.whenTerminated.futureValue shouldBe Done
}
diff --git a/tests/src/test/scala/docs/scaladsl/DocsSpecBase.scala b/tests/src/test/scala/docs/scaladsl/DocsSpecBase.scala
index a6603f2f..a8d97158 100644
--- a/tests/src/test/scala/docs/scaladsl/DocsSpecBase.scala
+++ b/tests/src/test/scala/docs/scaladsl/DocsSpecBase.scala
@@ -14,10 +14,11 @@
package docs.scaladsl
-import org.apache.pekko.NotUsed
-import org.apache.pekko.kafka.testkit.scaladsl.KafkaSpec
-import org.apache.pekko.kafka.testkit.internal.TestFrameworkInterface
-import org.apache.pekko.stream.scaladsl.Flow
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.kafka.testkit.scaladsl.KafkaSpec
+import pekko.kafka.testkit.internal.TestFrameworkInterface
+import pekko.stream.scaladsl.Flow
import org.scalatest.Suite
import org.scalatest.concurrent.{ Eventually, IntegrationPatience, ScalaFutures }
import org.scalatest.flatspec.AnyFlatSpecLike
diff --git a/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala b/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
index 9a9b7b44..1b447b26 100644
--- a/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
+++ b/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
@@ -14,17 +14,19 @@
package docs.scaladsl
-import org.apache.pekko.kafka.scaladsl.MetadataClient
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import org.apache.pekko
+import pekko.kafka.scaladsl.MetadataClient
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
import org.scalatest.TryValues
import org.scalatest.time.{ Seconds, Span }
// #metadata
// #metadataClient
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.kafka.{ KafkaConsumerActor, Metadata }
-import org.apache.pekko.pattern.ask
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.kafka.{ KafkaConsumerActor, Metadata }
+import pekko.pattern.ask
+import pekko.util.Timeout
import org.apache.kafka.common.TopicPartition
import scala.concurrent.Future
diff --git a/tests/src/test/scala/docs/scaladsl/PartitionExamples.scala b/tests/src/test/scala/docs/scaladsl/PartitionExamples.scala
index 1f2cc17f..1ebc513a 100644
--- a/tests/src/test/scala/docs/scaladsl/PartitionExamples.scala
+++ b/tests/src/test/scala/docs/scaladsl/PartitionExamples.scala
@@ -14,14 +14,15 @@
package docs.scaladsl
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.actor.typed.scaladsl.Behaviors
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.kafka.{ KafkaConsumerActor, Subscriptions }
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.kafka.scaladsl.Consumer
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.kafka.{ KafkaConsumerActor, Subscriptions }
+import pekko.stream.scaladsl.{ Keep, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
import scala.annotation.nowarn
@@ -95,7 +96,7 @@ class PartitionExamples extends DocsSpecBase with TestcontainersKafkaPerClassLik
val _ = Behaviors.setup[Nothing] { context =>
// #consumerActorTyped
// adds support for actors to a classic actor system and context
- import org.apache.pekko.actor.typed.scaladsl.adapter._
+ import pekko.actor.typed.scaladsl.adapter._
// Consumer is represented by actor
// #consumerActorTyped
diff --git a/tests/src/test/scala/docs/scaladsl/ProducerExample.scala b/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
index 86ec9452..bbe530ba 100644
--- a/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
@@ -14,13 +14,14 @@
package docs.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.ProducerMessage.MultiResultPart
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.kafka.{ ProducerMessage, ProducerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.ProducerMessage.MultiResultPart
+import pekko.kafka.scaladsl.{ Consumer, Producer }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.{ ProducerMessage, ProducerSettings, Subscriptions }
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer
diff --git a/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala b/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
index d93a9070..cd1612fa 100644
--- a/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
@@ -16,14 +16,15 @@ package docs.scaladsl
import java.nio.charset.StandardCharsets
-import org.apache.pekko.Done
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl._
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.scaladsl._
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig
import org.apache.avro.specific.SpecificRecordBase
import org.apache.avro.util.Utf8
diff --git a/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala b/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
index a8e33f90..0b15f9c4 100644
--- a/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
@@ -14,12 +14,13 @@
package docs.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.ProducerMessage.MultiResult
-import org.apache.pekko.kafka.scaladsl.{ Consumer, SendProducer }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.kafka.{ ConsumerSettings, ProducerMessage, Subscriptions }
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink }
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.ProducerMessage.MultiResult
+import pekko.kafka.scaladsl.{ Consumer, SendProducer }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.{ ConsumerSettings, ProducerMessage, Subscriptions }
+import pekko.stream.scaladsl.{ Keep, Sink }
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
import scala.collection.immutable
diff --git a/tests/src/test/scala/docs/scaladsl/SerializationSpec.scala b/tests/src/test/scala/docs/scaladsl/SerializationSpec.scala
index fcb7081d..d1a0947d 100644
--- a/tests/src/test/scala/docs/scaladsl/SerializationSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/SerializationSpec.scala
@@ -14,14 +14,15 @@
package docs.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.scaladsl._
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.{ ActorAttributes, Supervision }
-import org.apache.pekko.stream.scaladsl.{ Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.scaladsl._
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.{ ActorAttributes, Supervision }
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization._
diff --git a/tests/src/test/scala/docs/scaladsl/TestkitSamplesSpec.scala b/tests/src/test/scala/docs/scaladsl/TestkitSamplesSpec.scala
index f9dada1b..cccee797 100644
--- a/tests/src/test/scala/docs/scaladsl/TestkitSamplesSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/TestkitSamplesSpec.scala
@@ -14,14 +14,15 @@
package docs.scaladsl
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage.CommittableOffset
-import org.apache.pekko.kafka.scaladsl.{ Committer, Consumer }
-import org.apache.pekko.kafka.{ CommitterSettings, ConsumerMessage, ProducerMessage }
-import org.apache.pekko.stream.scaladsl.{ Flow, Keep, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.testkit.TestKit
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.CommittableOffset
+import pekko.kafka.scaladsl.{ Committer, Consumer }
+import pekko.kafka.{ CommitterSettings, ConsumerMessage, ProducerMessage }
+import pekko.stream.scaladsl.{ Flow, Keep, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.testkit.TestKit
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.producer.ProducerRecord
import org.scalatest.BeforeAndAfterAll
@@ -48,8 +49,9 @@ class TestkitSamplesSpec
val committerSettings = CommitterSettings(system)
// #factories
- import org.apache.pekko.kafka.testkit.scaladsl.ConsumerControlFactory
- import org.apache.pekko.kafka.testkit.{ ConsumerResultFactory, ProducerResultFactory }
+ import org.apache.pekko
+ import pekko.kafka.testkit.scaladsl.ConsumerControlFactory
+ import pekko.kafka.testkit.{ ConsumerResultFactory, ProducerResultFactory }
// create elements emitted by the mocked Consumer
val elements = (0 to 10).map { i =>
diff --git a/tests/src/test/scala/docs/scaladsl/TransactionsExample.scala b/tests/src/test/scala/docs/scaladsl/TransactionsExample.scala
index 82e04114..3e6cfbc4 100644
--- a/tests/src/test/scala/docs/scaladsl/TransactionsExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/TransactionsExample.scala
@@ -16,21 +16,15 @@ package docs.scaladsl
import java.util.concurrent.atomic.AtomicReference
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Transactional }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.kafka.{
- ConsumerSettings,
- ProducerMessage,
- ProducerSettings,
- Repeated,
- Subscriptions,
- TransactionsOps
-}
-import org.apache.pekko.stream.RestartSettings
-import org.apache.pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
+import pekko.kafka.scaladsl.{ Consumer, Transactional }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.{ ConsumerSettings, ProducerMessage, ProducerSettings, Repeated, Subscriptions, TransactionsOps }
+import pekko.stream.RestartSettings
+import pekko.stream.scaladsl.{ Keep, RestartSource, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import scala.concurrent.Await
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/ConfigSettingsSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/ConfigSettingsSpec.scala
index 74ac3703..7707065e 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/ConfigSettingsSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/ConfigSettingsSpec.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka
-import org.apache.pekko.kafka.internal.ConfigSettings
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
+import org.apache.pekko
+import pekko.kafka.internal.ConfigSettings
+import pekko.kafka.tests.scaladsl.LogCapturing
import com.typesafe.config.ConfigFactory
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala
index 79f20daa..76ebf20f 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.common.config.SslConfigs
import org.apache.kafka.common.serialization.{ ByteArrayDeserializer, StringDeserializer }
@@ -216,7 +217,7 @@ class ConsumerSettingsSpec
.resolve()
"read bootstrap servers from config" in {
- import org.apache.pekko.kafka.scaladsl.DiscoverySupport
+ import pekko.kafka.scaladsl.DiscoverySupport
implicit val actorSystem = ActorSystem("test", config)
DiscoverySupport.bootstrapServers(config.getConfig("discovery-consumer")).futureValue shouldBe "cat:1233,dog:1234"
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala
index 0c884935..198dd825 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.kafka
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.common.config.SslConfigs
import org.apache.kafka.common.serialization.{ ByteArraySerializer, StringSerializer }
@@ -220,7 +221,7 @@ class ProducerSettingsSpec
"fail if using non-async creation with enrichAsync" in {
implicit val actorSystem = ActorSystem("test", config)
- import org.apache.pekko.kafka.scaladsl.DiscoverySupport
+ import pekko.kafka.scaladsl.DiscoverySupport
val producerConfig = config.getConfig("discovery-producer")
val settings = ProducerSettings(producerConfig, new StringSerializer, new StringSerializer)
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala b/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
index db975254..4d7869b4 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
@@ -15,16 +15,17 @@
package org.apache.pekko.kafka
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.{ Done, NotUsed }
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage.PartitionOffset
-import org.apache.pekko.kafka.ProducerMessage.MultiMessage
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.{ Consumer, Producer, Transactional }
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.{ Flow, Sink, Source }
-import org.apache.pekko.stream.testkit.TestSubscriber
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
+import org.apache.pekko
+import pekko.{ Done, NotUsed }
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.PartitionOffset
+import pekko.kafka.ProducerMessage.MultiMessage
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.{ Consumer, Producer, Transactional }
+import pekko.stream.Materializer
+import pekko.stream.scaladsl.{ Flow, Sink, Source }
+import pekko.stream.testkit.TestSubscriber
+import pekko.stream.testkit.scaladsl.TestSink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{ ProducerConfig, ProducerRecord }
import org.scalatest.TestSuite
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommitCollectorStageSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommitCollectorStageSpec.scala
index 871733e1..0c8a3d66 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommitCollectorStageSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommitCollectorStageSpec.scala
@@ -15,20 +15,21 @@
package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicLong
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.kafka.ConsumerMessage.{ Committable, CommittableOffset, CommittableOffsetBatch }
-import org.apache.pekko.kafka.scaladsl.{ Committer, Consumer }
-import org.apache.pekko.kafka.testkit.ConsumerResultFactory
-import org.apache.pekko.kafka.testkit.scaladsl.{ ConsumerControlFactory, Slf4jToAkkaLoggingAdapter }
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ CommitWhen, CommitterSettings, Repeated }
-import org.apache.pekko.stream.scaladsl.Keep
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
-import org.apache.pekko.stream.testkit.{ TestPublisher, TestSubscriber }
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.event.LoggingAdapter
+import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffset, CommittableOffsetBatch }
+import pekko.kafka.scaladsl.{ Committer, Consumer }
+import pekko.kafka.testkit.ConsumerResultFactory
+import pekko.kafka.testkit.scaladsl.{ ConsumerControlFactory, Slf4jToAkkaLoggingAdapter }
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ CommitWhen, CommitterSettings, Repeated }
+import pekko.stream.scaladsl.Keep
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
+import pekko.stream.testkit.{ TestPublisher, TestSubscriber }
+import pekko.testkit.TestKit
import org.apache.kafka.clients.consumer.OffsetAndMetadata
import org.apache.kafka.common.TopicPartition
import org.scalatest.concurrent.{ Eventually, IntegrationPatience, ScalaFutures }
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkSpec.scala
index 77800e4b..367d1a8f 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkSpec.scala
@@ -16,20 +16,21 @@ package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicLong
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.scaladsl.Producer
-import org.apache.pekko.kafka.testkit.ConsumerResultFactory
-import org.apache.pekko.kafka.testkit.scaladsl.{ ConsumerControlFactory, Slf4jToAkkaLoggingAdapter }
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka._
-import org.apache.pekko.stream.scaladsl.{ Keep, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.{ ActorAttributes, Supervision }
-import org.apache.pekko.testkit.{ TestKit, TestProbe }
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.event.LoggingAdapter
+import pekko.kafka.internal.KafkaConsumerActor.Internal
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.scaladsl.Producer
+import pekko.kafka.testkit.ConsumerResultFactory
+import pekko.kafka.testkit.scaladsl.{ ConsumerControlFactory, Slf4jToAkkaLoggingAdapter }
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka._
+import pekko.stream.scaladsl.{ Keep, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.{ ActorAttributes, Supervision }
+import pekko.testkit.{ TestKit, TestProbe }
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.producer._
import org.apache.kafka.common.TopicPartition
@@ -469,7 +470,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
eventually {
producer.history.asScala should have size 2
}
- control.drainAndShutdown().failed.futureValue shouldBe an[org.apache.pekko.kafka.CommitTimeoutException]
+ control.drainAndShutdown().failed.futureValue shouldBe an[pekko.kafka.CommitTimeoutException]
}
it should "choose to ignore producer errors" in assertAllStagesStopped {
@@ -591,7 +592,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
eventually {
producer.history.asScala should have size 2
}
- control.drainAndShutdown().failed.futureValue shouldBe an[org.apache.pekko.kafka.CommitTimeoutException]
+ control.drainAndShutdown().failed.futureValue shouldBe an[pekko.kafka.CommitTimeoutException]
}
it should "ignore commit timeout" in assertAllStagesStopped {
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingWithMockSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingWithMockSpec.scala
index b705b1b1..9ede79de 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingWithMockSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/CommittingWithMockSpec.scala
@@ -16,18 +16,19 @@ package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage._
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.scaladsl.{ Committer, Consumer }
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.scaladsl._
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage._
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.scaladsl.{ Committer, Consumer }
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.stream._
+import pekko.stream.scaladsl._
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer._
import org.apache.kafka.common.TopicPartition
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConnectionCheckerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConnectionCheckerSpec.scala
index 55b42d46..b988335e 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConnectionCheckerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConnectionCheckerSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.{ ActorRef, ActorSystem }
-import org.apache.pekko.kafka.Metadata
-import org.apache.pekko.kafka.ConnectionCheckerSettings
-import org.apache.pekko.kafka.KafkaConnectionFailed
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.actor.{ ActorRef, ActorSystem }
+import pekko.kafka.Metadata
+import pekko.kafka.ConnectionCheckerSettings
+import pekko.kafka.KafkaConnectionFailed
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.common.errors.TimeoutException
import org.scalatest.wordspec.AnyWordSpecLike
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerMock.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerMock.scala
index 679545a5..ae0d07f0 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerMock.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerMock.scala
@@ -16,8 +16,9 @@ package org.apache.pekko.kafka.internal
import java.util.concurrent.atomic.AtomicBoolean
-import org.apache.pekko.testkit.TestKit
-import org.apache.pekko.util.JavaDurationConverters._
+import org.apache.pekko
+import pekko.testkit.TestKit
+import pekko.util.JavaDurationConverters._
import org.apache.kafka.clients.consumer._
import org.apache.kafka.common.TopicPartition
import org.mockito.Mockito._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerResetProtectionSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerResetProtectionSpec.scala
index d5e6fa88..b5d5c9eb 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerResetProtectionSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerResetProtectionSpec.scala
@@ -14,13 +14,14 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.event.LoggingAdapter
-import org.apache.pekko.kafka.OffsetResetProtectionSettings
-import org.apache.pekko.kafka.internal.KafkaConsumerActor.Internal.Seek
-import org.apache.pekko.kafka.testkit.scaladsl.Slf4jToAkkaLoggingAdapter
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.testkit.{ ImplicitSender, TestKit }
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.event.LoggingAdapter
+import pekko.kafka.OffsetResetProtectionSettings
+import pekko.kafka.internal.KafkaConsumerActor.Internal.Seek
+import pekko.kafka.testkit.scaladsl.Slf4jToAkkaLoggingAdapter
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.testkit.{ ImplicitSender, TestKit }
import org.apache.kafka.clients.consumer.{ ConsumerRecord, ConsumerRecords }
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.header.internals.RecordHeaders
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerSpec.scala
index b62dba6b..af77a6ef 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/ConsumerSpec.scala
@@ -14,17 +14,18 @@
package org.apache.pekko.kafka.internal
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage._
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.kafka.scaladsl.Consumer.Control
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ CommitTimeoutException, ConsumerSettings, Repeated, Subscriptions }
-import org.apache.pekko.stream.scaladsl._
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage._
+import pekko.kafka.scaladsl.Consumer
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ CommitTimeoutException, ConsumerSettings, Repeated, Subscriptions }
+import pekko.stream.scaladsl._
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer._
import org.apache.kafka.common.serialization.StringDeserializer
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/PartitionedSourceSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/PartitionedSourceSpec.scala
index 32cc2ce0..01c43089 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/PartitionedSourceSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/PartitionedSourceSpec.scala
@@ -18,16 +18,17 @@ import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.{ CountDownLatch, TimeUnit }
import java.util.function.UnaryOperator
-import org.apache.pekko.Done
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage._
-import org.apache.pekko.kafka.scaladsl.Consumer
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ ConsumerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl._
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage._
+import pekko.kafka.scaladsl.Consumer
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ ConsumerSettings, Subscriptions }
+import pekko.stream.scaladsl._
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer._
import org.apache.kafka.common.TopicPartition
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/ProducerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/ProducerSpec.scala
index 77ba03bf..d61583b2 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/ProducerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/ProducerSpec.scala
@@ -15,18 +15,19 @@
package org.apache.pekko.kafka.internal
import java.util.concurrent.CompletableFuture
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ConsumerMessage.{ GroupTopicPartition, PartitionOffset, PartitionOffsetCommittedMarker }
-import org.apache.pekko.kafka.ProducerMessage._
-import org.apache.pekko.kafka.scaladsl.Producer
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ ConsumerMessage, ProducerMessage, ProducerSettings }
-import org.apache.pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
-import org.apache.pekko.stream.{ ActorAttributes, Supervision }
-import org.apache.pekko.testkit.TestKit
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.ConsumerMessage.{ GroupTopicPartition, PartitionOffset, PartitionOffsetCommittedMarker }
+import pekko.kafka.ProducerMessage._
+import pekko.kafka.scaladsl.Producer
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ ConsumerMessage, ProducerMessage, ProducerSettings }
+import pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
+import pekko.stream.{ ActorAttributes, Supervision }
+import pekko.testkit.TestKit
+import pekko.{ Done, NotUsed }
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer.{ ConsumerGroupMetadata, OffsetAndMetadata }
import org.apache.kafka.clients.producer._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/internal/SubscriptionsSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/internal/SubscriptionsSpec.scala
index 989e0726..f8e9b4ac 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/internal/SubscriptionsSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/internal/SubscriptionsSpec.scala
@@ -16,9 +16,10 @@ package org.apache.pekko.kafka.internal
import java.net.URLEncoder
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ Subscription, Subscriptions }
-import org.apache.pekko.util.ByteString
+import org.apache.pekko
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ Subscription, Subscriptions }
+import pekko.util.ByteString
import org.apache.kafka.common.TopicPartition
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
index 63bb15be..961820a6 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
@@ -18,9 +18,10 @@ import java.util
import java.util.concurrent.{ CompletionStage, Executor, Executors }
import java.util.concurrent.atomic.AtomicBoolean
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.internal.ConsumerControlAsJava
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.internal.ConsumerControlAsJava
+import pekko.kafka.tests.scaladsl.LogCapturing
import org.apache.kafka.common.{ Metric, MetricName }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
@@ -33,7 +34,7 @@ import scala.language.reflectiveCalls
object ControlSpec {
def createControl(stopFuture: Future[Done] = Future.successful(Done),
shutdownFuture: Future[Done] = Future.successful(Done)) = {
- val control = new org.apache.pekko.kafka.scaladsl.ControlSpec.ControlImpl(stopFuture, shutdownFuture)
+ val control = new pekko.kafka.scaladsl.ControlSpec.ControlImpl(stopFuture, shutdownFuture)
val wrapped = new ConsumerControlAsJava(control)
new Consumer.Control {
def shutdownCalled: AtomicBoolean = control.shutdownCalled
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittableSinkSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittableSinkSpec.scala
index 7d6eb884..11e3c1ce 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittableSinkSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittableSinkSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import scala.collection.immutable
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittingSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittingSpec.scala
index 1dc09318..ca311cfd 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittingSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CommittingSpec.scala
@@ -17,18 +17,19 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicInteger
import java.util.function.IntUnaryOperator
-import org.apache.pekko.actor.ActorRef
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableOffsetBatch, GroupTopicPartition }
-import org.apache.pekko.kafka.ProducerMessage.MultiMessage
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.internal.CommittableOffsetBatchImpl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.RestartSettings
-import org.apache.pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestProbe
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.kafka.ConsumerMessage.{ CommittableOffsetBatch, GroupTopicPartition }
+import pekko.kafka.ProducerMessage.MultiMessage
+import pekko.kafka._
+import pekko.kafka.internal.CommittableOffsetBatchImpl
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.RestartSettings
+import pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestProbe
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
import org.scalatest.Inside
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ConnectionCheckerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ConnectionCheckerSpec.scala
index 847744b2..ce67baee 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ConnectionCheckerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ConnectionCheckerSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
+import org.apache.pekko
+import pekko.kafka._
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.scaladsl.{ Keep, Sink }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ControlSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ControlSpec.scala
index 604eec68..26071a6e 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ControlSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ControlSpec.scala
@@ -16,9 +16,10 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicBoolean
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.tests.scaladsl.LogCapturing
import org.apache.kafka.common.{ Metric, MetricName }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.wordspec.AnyWordSpec
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/IntegrationSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/IntegrationSpec.scala
index 6b37c388..a945e11d 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/IntegrationSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/IntegrationSpec.scala
@@ -16,17 +16,18 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicLong
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.ConsumerMessage.CommittableOffsetBatch
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.pattern.ask
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestProbe
-import org.apache.pekko.util.Timeout
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.ConsumerMessage.CommittableOffsetBatch
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.pattern.ask
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestProbe
+import pekko.util.Timeout
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MetadataClientSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MetadataClientSpec.scala
index 4a9e0892..2e7bcfc8 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MetadataClientSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MetadataClientSpec.scala
@@ -14,8 +14,9 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.common.{ PartitionInfo, TopicPartition }
import scala.language.postfixOps
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredConsumerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredConsumerSpec.scala
index 0ede6918..0e5e2801 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredConsumerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredConsumerSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.kafka.{ ConsumerSettings, Subscriptions }
-import org.apache.pekko.stream.scaladsl.Sink
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.kafka.{ ConsumerSettings, Subscriptions }
+import pekko.stream.scaladsl.Sink
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.testkit.TestKit
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.scalatest.concurrent.{ Eventually, IntegrationPatience, ScalaFutures }
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredProducerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredProducerSpec.scala
index f642db9c..5b346a73 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredProducerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MisconfiguredProducerSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.kafka.ProducerSettings
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
-import org.apache.pekko.stream.scaladsl.Source
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.testkit.TestKit
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.kafka.ProducerSettings
+import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.stream.scaladsl.Source
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.testkit.TestKit
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer
import org.scalatest.concurrent.{ Eventually, IntegrationPatience, ScalaFutures }
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MultiConsumerSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MultiConsumerSpec.scala
index 00dbace1..99124d8c 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MultiConsumerSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/MultiConsumerSpec.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import scala.collection.immutable
import scala.concurrent.duration._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/PartitionedSourcesSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/PartitionedSourcesSpec.scala
index af8faa85..955bd397 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/PartitionedSourcesSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/PartitionedSourcesSpec.scala
@@ -18,15 +18,16 @@ import java.util.concurrent.{ CountDownLatch, TimeUnit }
import java.util.concurrent.atomic.{ AtomicBoolean, AtomicLong }
import java.util.function.LongBinaryOperator
-import org.apache.pekko.Done
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.scaladsl.Consumer.DrainingControl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.{ KillSwitches, OverflowStrategy }
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestProbe
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.DrainingControl
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.{ KillSwitches, OverflowStrategy }
+import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestProbe
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord }
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
@@ -562,7 +563,7 @@ class PartitionedSourcesSpec extends SpecBase with TestcontainersKafkaLike with
exceptionTriggered.set(true)
Future.failed(new RuntimeException("FAIL"))
} else {
- org.apache.pekko.pattern.after(50.millis, system.scheduler)(Future.successful(m))
+ pekko.pattern.after(50.millis, system.scheduler)(Future.successful(m))
}
}
.log(s"subsource $tp pre commit")
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceExtSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceExtSpec.scala
index def53693..97b1ce89 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceExtSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceExtSpec.scala
@@ -16,13 +16,14 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream._
-import org.apache.pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.kafka.ConsumerMessage.{ CommittableMessage, CommittableOffset }
+import pekko.kafka._
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream._
+import pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.TopicPartition
import org.scalatest._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
index ed882ad1..f39afe0f 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
@@ -17,14 +17,15 @@ package org.apache.pekko.kafka.scaladsl
import java.util
import java.util.concurrent.atomic.AtomicReference
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Source }
-import org.apache.pekko.stream.testkit.TestSubscriber
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
-import org.apache.pekko.testkit.TestProbe
-import org.apache.pekko.{ Done, NotUsed }
+import org.apache.pekko
+import pekko.kafka._
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Keep, Source }
+import pekko.stream.testkit.TestSubscriber
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestProbe
+import pekko.{ Done, NotUsed }
import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerPartitionAssignor, ConsumerRecord }
import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
import org.apache.kafka.common.TopicPartition
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ReconnectSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ReconnectSpec.scala
index 54b52f78..596e28d2 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ReconnectSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/ReconnectSpec.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.stream.scaladsl.{ Keep, Sink, Source, SourceQueueWithComplete, Tcp }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.{ KillSwitches, OverflowStrategy, UniqueKillSwitch }
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Keep, Sink, Source, SourceQueueWithComplete, Tcp }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.{ KillSwitches, OverflowStrategy, UniqueKillSwitch }
import org.apache.kafka.clients.producer.ProducerRecord
import scala.concurrent.duration._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
index de171b1e..116fb8f0 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
@@ -16,13 +16,14 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.ConcurrentLinkedQueue
-import org.apache.pekko.Done
-import org.apache.pekko.kafka._
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
-import org.apache.pekko.stream.scaladsl.Keep
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.stream.scaladsl.Keep
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
import scala.concurrent.Await
import scala.concurrent.duration._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/SpecBase.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/SpecBase.scala
index 1022d624..2684fb59 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/SpecBase.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/SpecBase.scala
@@ -14,10 +14,12 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.kafka.Repeated
-import org.apache.pekko.kafka.tests.scaladsl.LogCapturing
+import org.apache.pekko
+import pekko.kafka.Repeated
+import pekko.kafka.tests.scaladsl.LogCapturing
// #testkit
-import org.apache.pekko.kafka.testkit.scaladsl.ScalatestKafkaSpec
+import org.apache.pekko
+import pekko.kafka.testkit.scaladsl.ScalatestKafkaSpec
import org.scalatest.concurrent.{ Eventually, IntegrationPatience, ScalaFutures }
import org.scalatest.wordspec.AnyWordSpecLike
import org.scalatest.matchers.should.Matchers
@@ -40,7 +42,7 @@ abstract class SpecBase(kafkaPort: Int)
// #testkit
// #testcontainers
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
class TestcontainersSampleSpec extends SpecBase with TestcontainersKafkaLike {
// ...
@@ -48,8 +50,8 @@ class TestcontainersSampleSpec extends SpecBase with TestcontainersKafkaLike {
// #testcontainers
// #testcontainers-settings
-import org.apache.pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
+import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike
class TestcontainersNewSettingsSampleSpec extends SpecBase with TestcontainersKafkaPerClassLike {
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TimestampSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TimestampSpec.scala
index 67288211..9eb039eb 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TimestampSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TimestampSpec.scala
@@ -14,10 +14,11 @@
package org.apache.pekko.kafka.scaladsl
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.kafka.Subscriptions
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import org.apache.pekko.stream.testkit.scaladsl.TestSink
+import org.apache.pekko
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.Subscriptions
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
import org.apache.kafka.common.TopicPartition
import org.scalatest.Inside
import org.scalatest.concurrent.IntegrationPatience
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TransactionsSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TransactionsSpec.scala
index d49ce3e0..e97ef021 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TransactionsSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/TransactionsSpec.scala
@@ -16,14 +16,15 @@ package org.apache.pekko.kafka.scaladsl
import java.util.concurrent.atomic.AtomicBoolean
-import org.apache.pekko.Done
-import org.apache.pekko.kafka.ConsumerMessage.PartitionOffset
-import org.apache.pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
-import org.apache.pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
-import org.apache.pekko.kafka.{ ProducerMessage, _ }
-import org.apache.pekko.stream.{ OverflowStrategy, RestartSettings }
-import org.apache.pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
-import org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka.ConsumerMessage.PartitionOffset
+import pekko.kafka.scaladsl.Consumer.{ Control, DrainingControl }
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.kafka.{ ProducerMessage, _ }
+import pekko.stream.{ OverflowStrategy, RestartSettings }
+import pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerRecord
import org.scalatest.RecoverMethods._
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/tests/CapturingAppender.scala b/tests/src/test/scala/org/apache/pekko/kafka/tests/CapturingAppender.scala
index f3a33d1e..a89a503a 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/tests/CapturingAppender.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/tests/CapturingAppender.scala
@@ -14,7 +14,8 @@
package org.apache.pekko.kafka.tests
-import org.apache.pekko.annotation.InternalApi
+import org.apache.pekko
+import pekko.annotation.InternalApi
import ch.qos.logback.classic.spi.ILoggingEvent
import ch.qos.logback.core.AppenderBase
@@ -49,10 +50,10 @@ import ch.qos.logback.core.AppenderBase
*
* Logging from tests can be silenced by this appender. When there is a test failure
* the captured logging events are flushed to the appenders defined for the
- * org.apache.org.apache.pekko.actor.testkit.typed.internal.CapturingAppenderDelegate logger.
+ * pekko.actor.testkit.typed.internal.CapturingAppenderDelegate logger.
*
- * The flushing on test failure is handled by [[org.apache.pekko.actor.testkit.typed.scaladsl.LogCapturing]]
- * for ScalaTest and [[org.apache.pekko.actor.testkit.typed.javadsl.LogCapturing]] for JUnit.
+ * The flushing on test failure is handled by [[pekko.actor.testkit.typed.scaladsl.LogCapturing]]
+ * for ScalaTest and [[pekko.actor.testkit.typed.javadsl.LogCapturing]] for JUnit.
*
* Use configuration like the following the logback-test.xml:
*
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org