This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 28ccd8d7 Replace immutable.Seq with Seq (#655)
28ccd8d7 is described below
commit 28ccd8d765ec3600f37767ff133ce09b40f5e50f
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Sep 13 08:31:23 2026 +0100
Replace immutable.Seq with Seq (#655)
Motivation:
The codebase still carried Scala 2.12-era `immutable.Seq` qualifiers.
Since Scala 2.13, `scala.Seq` is a type alias for
`scala.collection.immutable.Seq`, so the qualifier is redundant.
Equivalent of apache/pekko#3539.
Modification:
- Replace `immutable.Seq` with `Seq` in main, testkit, test and doc
sources and drop the now-unused `scala.collection.immutable` imports.
- Files that still use other `immutable.*` types keep the import.
Result:
No remaining `immutable.Seq` qualifiers. Source and binary compatible:
`Seq` erases to the same `scala.collection.immutable.Seq` class.
Tests:
- native `scalafmt` run on all changed files (no changes)
- `sbt Test/compile` passes
- `sbt core/mimaReportBinaryIssues testkit/mimaReportBinaryIssues` pass
- `sbt "tests/testOnly ...CommittingProducerSinkSpec ...ConsumerSpec"` pass
References:
None - Scala 2.12 support was dropped previously; port of apache/pekko#3539
---
.../org/apache/pekko/kafka/ProducerMessage.scala | 9 +++---
.../kafka/TransactionsPartitionedSourceSpec.scala | 3 +-
.../pekko/kafka/TransactionsSourceSpec.scala | 3 +-
.../kafka/testkit/ProducerResultFactory.scala | 3 +-
.../pekko/kafka/testkit/scaladsl/KafkaSpec.scala | 11 ++++----
.../test/scala/docs/scaladsl/AssignmentSpec.scala | 4 +--
.../src/test/scala/docs/scaladsl/AtLeastOnce.scala | 7 ++---
.../test/scala/docs/scaladsl/ConsumerExample.scala | 3 +-
.../test/scala/docs/scaladsl/ProducerExample.scala | 3 +-
.../scaladsl/SchemaRegistrySerializationSpec.scala | 5 ++--
.../scala/docs/scaladsl/SendProducerSpec.scala | 9 +++---
.../org/apache/pekko/kafka/TransactionsOps.scala | 12 ++++----
.../internal/CommittingProducerSinkSpec.scala | 33 +++++++++++-----------
.../apache/pekko/kafka/internal/ConsumerMock.scala | 1 -
.../apache/pekko/kafka/internal/ConsumerSpec.scala | 1 -
.../pekko/kafka/scaladsl/CommittableSinkSpec.scala | 3 +-
.../pekko/kafka/scaladsl/CommittingSpec.scala | 9 +++---
.../kafka/scaladsl/ConnectionCheckerSpec.scala | 2 +-
.../pekko/kafka/scaladsl/MultiConsumerSpec.scala | 3 +-
.../pekko/kafka/scaladsl/TransactionsSpec.scala | 6 ++--
20 files changed, 56 insertions(+), 74 deletions(-)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ProducerMessage.scala
b/core/src/main/scala/org/apache/pekko/kafka/ProducerMessage.scala
index 0690403f..fde84382 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ProducerMessage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ProducerMessage.scala
@@ -18,7 +18,6 @@ import org.apache.pekko
import pekko.NotUsed
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
/**
@@ -100,7 +99,7 @@ object ProducerMessage {
* that can be committed later in the flow.
*/
final case class MultiMessage[K, V, +PassThrough](
- records: immutable.Seq[ProducerRecord[K, V]],
+ records: Seq[ProducerRecord[K, V]],
passThrough: PassThrough) extends Envelope[K, V, PassThrough] {
/**
@@ -123,7 +122,7 @@ object ProducerMessage {
* @tparam PassThrough the type of data passed through
*/
def multi[K, V, PassThrough](
- records: immutable.Seq[ProducerRecord[K, V]],
+ records: Seq[ProducerRecord[K, V]],
passThrough: PassThrough): Envelope[K, V, PassThrough] =
MultiMessage(records, passThrough)
/**
@@ -133,7 +132,7 @@ object ProducerMessage {
* @tparam V the type of values
*/
def multi[K, V](
- records: immutable.Seq[ProducerRecord[K, V]]): Envelope[K, V, NotUsed] =
MultiMessage(records, NotUsed)
+ records: Seq[ProducerRecord[K, V]]): Envelope[K, V, NotUsed] =
MultiMessage(records, NotUsed)
/**
* Java API:
@@ -222,7 +221,7 @@ object ProducerMessage {
* successfully published.
*/
final case class MultiResult[K, V, PassThrough] private[kafka] (
- parts: immutable.Seq[MultiResultPart[K, V]],
+ parts: Seq[MultiResultPart[K, V]],
passThrough: PassThrough) extends Results[K, V, PassThrough] {
/**
diff --git
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
index 22d6d35b..fd48ca12 100644
---
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
+++
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsPartitionedSourceSpec.scala
@@ -30,7 +30,6 @@ import org.scalatest.Ignore
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, TimeoutException }
import scala.util.{ Failure, Success }
@@ -75,7 +74,7 @@ class TransactionsPartitionedSourceSpec
val elements = 100 * 1000 // 100 * 1,000 = 100,000
val restartAfter = (10 * 1000) / sourcePartitions // (10 * 1,000) / 10 =
100
- val producers: immutable.Seq[Future[Done]] = (0 until
sourcePartitions).map { part =>
+ val producers: Seq[Future[Done]] = (0 until sourcePartitions).map { part
=>
produce(sourceTopic, range = 1 to elements, partition = part)
}
diff --git
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
index 4d9b6a60..67d5ac36 100644
---
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
+++
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
@@ -30,7 +30,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, TimeoutException }
import scala.util.{ Failure, Success }
@@ -71,7 +70,7 @@ class TransactionsSourceSpec
val restartAfter = 10 * 1000
val partitionSize = elements / sourcePartitions
- val producers: immutable.Seq[Future[Done]] = (0 until
sourcePartitions).map { part =>
+ val producers: Seq[Future[Done]] = (0 until sourcePartitions).map { part
=>
val rangeStart = (part * partitionSize) + 1
val rangeEnd = partitionSize * (part + 1)
log.info(s"Producing [$rangeStart to $rangeEnd] to partition $part")
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 12986189..9da9215b 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
@@ -20,7 +20,6 @@ import pekko.kafka.ProducerMessage
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
import org.apache.kafka.common.TopicPartition
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
/**
@@ -54,7 +53,7 @@ object ProducerResultFactory {
ProducerMessage.MultiResultPart(metadata, record)
def multiResult[K, V, PassThrough](
- parts: immutable.Seq[ProducerMessage.MultiResultPart[K, V]],
+ parts: Seq[ProducerMessage.MultiResultPart[K, V]],
passThrough: PassThrough): ProducerMessage.MultiResult[K, V,
PassThrough] =
ProducerMessage.MultiResult(parts, passThrough)
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 a47ef989..673d9d0b 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
@@ -36,7 +36,6 @@ import org.apache.kafka.clients.producer.{ Producer =>
KProducer, ProducerRecord
import org.apache.kafka.common.GroupState
import org.slf4j.{ Logger, LoggerFactory }
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.jdk.CollectionConverters._
@@ -125,7 +124,7 @@ abstract class KafkaSpec(_kafkaPort: Int, val
zooKeeperPort: Int, actorSystem: A
Try(predicate(group.members().asScala.toList)).getOrElse(false)
}
- def createTopics(topics: Int*): immutable.Seq[String] = {
+ def createTopics(topics: Int*): Seq[String] = {
val topicNames = topics.toList.map { number =>
createTopicName(number)
}
@@ -146,10 +145,10 @@ abstract class KafkaSpec(_kafkaPort: Int, val
zooKeeperPort: Int, actorSystem: A
* Produce messages to topic using specified range and return
* a Future so the caller can synchronize consumption.
*/
- def produce(topic: String, range: immutable.Seq[Int], partition: Int =
partition0): Future[Done] =
+ def produce(topic: String, range: Seq[Int], partition: Int = partition0):
Future[Done] =
produceString(topic, range.map(_.toString), partition)
- def produceString(topic: String, range: immutable.Seq[String], partition:
Int = partition0): Future[Done] =
+ def produceString(topic: String, range: Seq[String], partition: Int =
partition0): Future[Done] =
Source(range)
// NOTE: If no partition is specified but a key is present a partition
will be chosen
// using a hash of the key. If neither key nor partition is present a
partition
@@ -166,7 +165,7 @@ abstract class KafkaSpec(_kafkaPort: Int, val
zooKeeperPort: Int, actorSystem: A
.map(n => new ProducerRecord(topic, partition0, DefaultKey, n.toString))
.runWith(Producer.plainSink(settings))
- def produceTimestamped(topic: String, timestampedRange: immutable.Seq[(Int,
Long)]): Future[Done] =
+ def produceTimestamped(topic: String, timestampedRange: Seq[(Int, Long)]):
Future[Done] =
Source(timestampedRange)
.map {
case (n, ts) => new ProducerRecord(topic, partition0, ts, DefaultKey,
n.toString)
@@ -177,7 +176,7 @@ abstract class KafkaSpec(_kafkaPort: Int, val
zooKeeperPort: Int, actorSystem: A
* Produce batches over several topics.
*/
def produceBatches(topics: Seq[String], batches: Int, batchSize: Int):
Future[Seq[Done]] = {
- val produceMessages: immutable.Seq[Future[Done]] = (0 until batches)
+ val produceMessages: Seq[Future[Done]] = (0 until batches)
.flatMap { batch =>
topics.map { topic =>
val batchStart = batch * batchSize
diff --git a/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
b/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
index a45ee3c3..c21b82f7 100644
--- a/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/AssignmentSpec.scala
@@ -24,8 +24,6 @@ import
pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
-import scala.collection.immutable
-
class AssignmentSpec extends SpecBase with TestcontainersKafkaLike {
"subscription with partition assignment" must {
@@ -56,7 +54,7 @@ class AssignmentSpec extends SpecBase with
TestcontainersKafkaLike {
"consume from the specified topic pattern" in assertAllStagesStopped {
val suffix = (System.currentTimeMillis() % 10000).toInt
- val topics = immutable.Seq(createTopic(suffix), createTopic(suffix))
+ val topics = Seq(createTopic(suffix), createTopic(suffix))
val group = createGroupId()
val totalMessages = 100
val producerCompletion =
diff --git a/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
b/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
index 3aac068b..98fc5018 100644
--- a/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
+++ b/tests/src/test/scala/docs/scaladsl/AtLeastOnce.scala
@@ -27,7 +27,6 @@ import pekko.stream.scaladsl.{ Keep, Sink }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
-import scala.collection.immutable
import scala.concurrent.Await
import scala.concurrent.duration._
@@ -50,7 +49,7 @@ class AtLeastOnce extends DocsSpecBase with
TestcontainersKafkaLike {
.committableSource(consumerSettings, Subscriptions.topics(topic1))
.map(msg =>
ProducerMessage.multi(
- immutable.Seq(
+ Seq(
new ProducerRecord(topic2, msg.record.key, msg.record.value),
new ProducerRecord(topic3, msg.record.key, msg.record.value)),
msg.committableOffset))
@@ -90,7 +89,7 @@ class AtLeastOnce extends DocsSpecBase with
TestcontainersKafkaLike {
val out: Envelope[String, String, CommittableOffset] =
if (duplicate(msg.record.value))
ProducerMessage.multi(
- immutable.Seq(
+ Seq(
new ProducerRecord(topic2, msg.record.key, msg.record.value),
new ProducerRecord(topic3, msg.record.key,
msg.record.value)),
msg.committableOffset)
@@ -138,7 +137,7 @@ class AtLeastOnce extends DocsSpecBase with
TestcontainersKafkaLike {
val out: Envelope[String, String, NotUsed] =
if (duplicate(record.value))
ProducerMessage.multi(
- immutable.Seq(
+ Seq(
new ProducerRecord(topic2, record.key, record.value),
new ProducerRecord(topic3, record.key, record.value)))
else if (ignore(record.value))
diff --git a/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
b/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
index 70c099e3..e0af3a79 100644
--- a/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/ConsumerExample.scala
@@ -33,7 +33,6 @@ import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.{ ByteArrayDeserializer,
StringDeserializer }
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, Promise }
@@ -135,7 +134,7 @@ class ConsumerExample extends DocsSpecBase with
TestcontainersKafkaLike {
}
// #atMostOnce
- val control: DrainingControl[immutable.Seq[Done]] =
+ val control: DrainingControl[Seq[Done]] =
Consumer
.atMostOnceSource(consumerSettings, Subscriptions.topics(topic))
.mapAsync(1)(record => business(record.key, record.value()))
diff --git a/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
b/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
index a068a0e2..e8f1afc7 100644
--- a/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
+++ b/tests/src/test/scala/docs/scaladsl/ProducerExample.scala
@@ -140,11 +140,10 @@ class ProducerExample extends DocsSpecBase with
TestcontainersKafkaLike {
key: KeyType,
value: ValueType,
passThrough: PassThroughType): ProducerMessage.Envelope[KeyType,
ValueType, PassThroughType] = {
- import scala.collection.immutable
// #multiMessage
val multi: ProducerMessage.Envelope[KeyType, ValueType, PassThroughType] =
ProducerMessage.multi(
- immutable.Seq(
+ Seq(
new ProducerRecord("topicName", key, value),
new ProducerRecord("anotherTopic", key, value)),
passThrough)
diff --git
a/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
b/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
index bccd1d65..06e49d9a 100644
--- a/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/SchemaRegistrySerializationSpec.scala
@@ -31,7 +31,6 @@ import org.apache.avro.util.Utf8
import org.apache.avro.{ AvroRuntimeException, Schema }
import org.apache.kafka.common.TopicPartition
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
// #imports
@@ -116,7 +115,7 @@ class SchemaRegistrySerializationSpec extends DocsSpecBase
with TestcontainersKa
ProducerSettings(system, new StringSerializer, new ByteArraySerializer)
.withBootstrapServers(bootstrapServers)
- val samples = immutable.Seq("String1")
+ val samples = Seq("String1")
val producerCompletion =
Source(samples)
.map(n => new ProducerRecord(topic, n,
n.getBytes(StandardCharsets.UTF_8)))
@@ -141,7 +140,7 @@ class SchemaRegistrySerializationSpec extends DocsSpecBase
with TestcontainersKa
val consumerActor =
system.actorOf(KafkaConsumerActor.props(specificRecordConsumerSettings(group)),
"sharedKafkaConsumerActor")
- val samples = immutable.Seq("String1", "String2", "String3")
+ val samples = Seq("String1", "String2", "String3")
val (control1, probe1) =
Consumer
diff --git a/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
b/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
index 07e2edbf..422f5541 100644
--- a/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
+++ b/tests/src/test/scala/docs/scaladsl/SendProducerSpec.scala
@@ -23,7 +23,6 @@ import pekko.kafka.{ ConsumerSettings, ProducerMessage,
Subscriptions }
import pekko.stream.scaladsl.{ Keep, Sink }
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future }
@@ -57,14 +56,14 @@ class SendProducerSpec extends DocsSpecBase with
TestcontainersKafkaLike {
val producer = SendProducer(producerDefaults)
try {
// #envelope
- val message = ProducerMessage.multi(immutable.Seq(new
ProducerRecord(topic1, "key", "value")), "context")
+ val message = ProducerMessage.multi(Seq(new ProducerRecord(topic1,
"key", "value")), "context")
val send: Future[ProducerMessage.Results[String, String, String]] =
producer.sendEnvelope(message)
// Blocking here for illustration only, you need to handle the future
result
Await.result(send, 2.seconds)
// #envelope
val result = send.futureValue
result match {
- case MultiResult(immutable.Seq(part), "context") =>
+ case MultiResult(Seq(part), "context") =>
part.metadata.topic() shouldBe topic1
case other => fail(s"unexpected result $other")
}
@@ -83,7 +82,7 @@ class SendProducerSpec extends DocsSpecBase with
TestcontainersKafkaLike {
val producer = SendProducer(producerDefaults)
try {
val envelope: ProducerMessage.Envelope[String, String, String] =
- ProducerMessage.multi(immutable.Seq(
+ ProducerMessage.multi(Seq(
new ProducerRecord(topic1, "key", "value1"),
new ProducerRecord(topic1, "key", "value2"),
new ProducerRecord(topic1, "key", "value3")),
@@ -92,7 +91,7 @@ class SendProducerSpec extends DocsSpecBase with
TestcontainersKafkaLike {
// #multiMessage
val result = send.futureValue
result match {
- case MultiResult(immutable.Seq(part1, part2, part3), "context") =>
+ case MultiResult(Seq(part1, part2, part3), "context") =>
part1.metadata.topic() shouldBe topic1
case other => fail(s"unexpected result $other")
}
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 8d4ec3c6..4f273e2b 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
@@ -122,7 +122,7 @@ trait TransactionsOps extends TestSuite with Matchers {
.via(Producer.flexiFlow(producerSettings))
.runWith(Sink.ignore)
- def checkForDuplicates(values: immutable.Seq[(Long, String)], expected:
immutable.IndexedSeq[String]): Unit =
+ def checkForDuplicates(values: Seq[(Long, String)], expected:
immutable.IndexedSeq[String]): Unit =
withClue("Checking for duplicates: ") {
val duplicates = values.map(_._2).diff(expected)
if (duplicates.nonEmpty) {
@@ -147,7 +147,7 @@ trait TransactionsOps extends TestSuite with Matchers {
}
}
- def checkForMissing(values: immutable.Seq[(Long, String)], expected:
immutable.IndexedSeq[String]): Unit =
+ def checkForMissing(values: Seq[(Long, String)], expected:
immutable.IndexedSeq[String]): Unit =
withClue("Checking for missing: ") {
val missing = expected.diff(values.map(_._2))
if (missing.nonEmpty) {
@@ -181,7 +181,7 @@ trait TransactionsOps extends TestSuite with Matchers {
.map(r => (r.offset(), r.value()))
def consumePartitionOffsetValues(settings: ConsumerSettings[String, String],
topic: String, elementsToTake: Long)(
- implicit mat: Materializer): Future[immutable.Seq[(Int, Long, String)]] =
+ implicit mat: Materializer): Future[Seq[(Int, Long, String)]] =
Consumer
.plainSource(settings, Subscriptions.topics(topic))
.map(r => (r.partition(), r.offset(), r.value()))
@@ -201,13 +201,13 @@ trait TransactionsOps extends TestSuite with Matchers {
def assertPartitionedConsistency(
elements: Int,
maxPartitions: Int,
- values: immutable.Seq[(Int, Long, String)]): Unit = {
- val expectedValues: immutable.Seq[String] = (1 to elements).map(_.toString)
+ values: Seq[(Int, Long, String)]): Unit = {
+ val expectedValues: Seq[String] = (1 to elements).map(_.toString)
for (partition <- 0 until maxPartitions) {
println(s"Asserting values for partition: $partition")
- val partitionMessages: immutable.Seq[String] =
+ val partitionMessages: Seq[String] =
values.filter(_._1 == partition).map { case (_, _, value) => value }
assert(partitionMessages.length == elements)
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 8ba35d38..722bc0ef 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
@@ -40,7 +40,6 @@ import org.scalatest.flatspec.AnyFlatSpecLike
import org.scalatest.matchers.should.Matchers
import org.slf4j.{ Logger, LoggerFactory }
-import scala.collection.immutable
import scala.concurrent.ExecutionContext
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -77,7 +76,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
"committable producer sink" should "produce, and commit after interval" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -113,7 +112,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit after interval with pass-through messages" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "skip"),
consumer.message(partition, "send"))
@@ -153,7 +152,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit when batch size is reached" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -187,7 +186,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit when batch size is reached with pass-through
messages" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -219,7 +218,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit when batch size is reached with
multi-messages" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -269,7 +268,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
.concat(Source.maybe)
.viaMat(ConsumerControlFactory.controlFlow())(Keep.right)
.map { msg =>
- ProducerMessage.multi(immutable.Seq.empty[ProducerRecord[String,
String]], msg.committableOffset)
+ ProducerMessage.multi(Seq.empty[ProducerRecord[String, String]],
msg.committableOffset)
}
.toMat(Producer.committableSink(producerSettings,
committerSettings))(DrainingControl.apply)
.run()
@@ -288,7 +287,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit when the next offset is observed" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -324,7 +323,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit on completion" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -360,7 +359,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit on delayed completion" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -400,7 +399,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "produce, and commit on upstream failure" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -436,7 +435,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "time out for missing producer reply" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -476,7 +475,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "choose to ignore producer errors" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -519,7 +518,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "choose to ignore producer errors and shut down cleanly" in
assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -565,7 +564,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "fail for commit timeout" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -598,7 +597,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "ignore commit timeout" in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
@@ -636,7 +635,7 @@ class CommittingProducerSinkSpec(_system: ActorSystem)
it should "not commit next offset after failure if it hasn't been observed"
in assertAllStagesStopped {
val consumer = FakeConsumer(groupId, topic, startOffset = 1616L)
- val elements = immutable.Seq(
+ val elements = Seq(
consumer.message(partition, "value 1"),
consumer.message(partition, "value 2"))
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 99c9c855..73209562 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
@@ -26,7 +26,6 @@ import org.mockito.stubbing.Answer
import org.mockito.verification.VerificationMode
import org.mockito.{ ArgumentMatchers, Mockito }
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
import scala.jdk.DurationConverters._
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 12b32977..70fc8799 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
@@ -39,7 +39,6 @@ import org.scalatest.concurrent.Eventually
import org.scalatest.flatspec.AnyFlatSpecLike
import org.scalatest.matchers.should.Matchers
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.jdk.CollectionConverters._
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 67de2480..28e44895 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
@@ -23,7 +23,6 @@ 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
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -111,7 +110,7 @@ class CommittableSinkSpec extends SpecBase with
TestcontainersKafkaLike {
}
}
- def produceStringRoundRobin(topic: String, range: immutable.Seq[String]):
Future[Done] =
+ def produceStringRoundRobin(topic: String, range: Seq[String]): Future[Done]
=
Source(range)
// NOTE: If no partition is specified but a key is present a partition
will be chosen
// using a hash of the key. If neither key nor partition is present a
partition
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 722bbe89..155564ae 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
@@ -34,7 +34,6 @@ import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
import org.scalatest.Inside
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -140,7 +139,7 @@ class CommittingSpec extends SpecBase with
TestcontainersKafkaLike with Inside {
Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
// read all messages from both partitions
- val committables1:
immutable.Seq[ConsumerMessage.CommittableMessage[String, String]] = probe1
+ val committables1: Seq[ConsumerMessage.CommittableMessage[String,
String]] = probe1
.request(count * 2L)
.expectNextN(count * 2L)
@@ -220,7 +219,7 @@ class CommittingSpec extends SpecBase with
TestcontainersKafkaLike with Inside {
Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
// read all messages from both partitions
- val committables1:
immutable.Seq[ConsumerMessage.CommittableMessage[String, String]] = probe1
+ val committables1: Seq[ConsumerMessage.CommittableMessage[String,
String]] = probe1
.request(count * 2L)
.expectNextN(count * 2L)
@@ -248,7 +247,7 @@ class CommittingSpec extends SpecBase with
TestcontainersKafkaLike with Inside {
}
})
- val committables2:
immutable.Seq[ConsumerMessage.CommittableMessage[String, String]] = probe2
+ val committables2: Seq[ConsumerMessage.CommittableMessage[String,
String]] = probe2
.request(count.toLong)
.expectNextN(count.toLong)
@@ -596,7 +595,7 @@ class CommittingSpec extends SpecBase with
TestcontainersKafkaLike with Inside {
Source(1 to 10)
.map(_.toString)
.mapConcat(n =>
- immutable.Seq(new ProducerRecord(topic, partition0, DefaultKey, n),
+ Seq(new ProducerRecord(topic, partition0, DefaultKey, n),
new ProducerRecord(topic, partition1, DefaultKey, n)))
.runWith(Producer.plainSink(producerDefaults.withProducer(testProducer)))
}
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 217a7e21..20fcb519 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
@@ -79,7 +79,7 @@ class ConnectionCheckerSpec extends SpecBase with
TestcontainersKafkaPerClassLik
startCluster()
val msg = "hello"
- produceString(topic, scala.collection.immutable.Seq(msg))
+ produceString(topic, Seq(msg))
val consumerSettings =
noBrokerConsumerSettings.withBootstrapServers(bootstrapServers)
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 cca5756f..ea774422 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
@@ -20,7 +20,6 @@ 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._
import scala.concurrent.{ Await, Future }
@@ -111,7 +110,7 @@ class MultiConsumerSpec extends SpecBase with
TestcontainersKafkaPerClassLike {
// produce 10 batches of 10 elements to all topics on different
partitions
val batches = 10
val batchSize = 10
- val produceMessages: immutable.Seq[Future[Done]] = (0 until batches)
+ val produceMessages: Seq[Future[Done]] = (0 until batches)
.flatMap { batch =>
topics.map { topic =>
val batchStart = batch * batchSize
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 be8cadb1..818615d0 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
@@ -269,7 +269,7 @@ class TransactionsSpec extends SpecBase with
TestcontainersKafkaLike with Transa
val concat = msgs.map(_.record.value()).reduce(_ + _)
ProducerMessage.multi(
- immutable.Seq(
+ Seq(
new ProducerRecord[String, String](sumsTopic, sum),
new ProducerRecord[String, String](concatsTopic, concat)),
msgs.map(_.partitionOffset).maxBy(_.offset))
@@ -344,7 +344,7 @@ class TransactionsSpec extends SpecBase with
TestcontainersKafkaLike with Transa
sinkTopic,
elementsToTake = (elements * maxPartitions).toLong)
- val actualValues: immutable.Seq[(Int, Long, String)] =
Await.result(consumer, 60.seconds)
+ val actualValues: Seq[(Int, Long, String)] = Await.result(consumer,
60.seconds)
assertPartitionedConsistency(elements, maxPartitions, actualValues)
Await.result(control.shutdown(), remainingOrDefault)
@@ -412,7 +412,7 @@ class TransactionsSpec extends SpecBase with
TestcontainersKafkaLike with Transa
elementsToTake = (elements * maxPartitions).toLong)
log.info("Retrieve actual values")
- val actualValues: immutable.Seq[(Int, Long, String)] =
Await.result(consumer, 60.seconds)
+ val actualValues: Seq[(Int, Long, String)] = Await.result(consumer,
60.seconds)
log.info("Waiting until partitions are assigned to one non-failed
consumer")
waitUntilConsumerSummary(group) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
