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 e5307adb Update pekkoVersion to 2.0.0-M4 (#643)
e5307adb is described below
commit e5307adbed942ea2a3265aeb40e8e464dee61e65
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Aug 22 12:47:02 2026 +0100
Update pekkoVersion to 2.0.0-M4 (#643)
* Update pekkoVersion to 2.0.0-M4
* try to avoid the deprecated method
---
project/PekkoCoreDependency.scala | 2 +-
.../pekko/kafka/scaladsl/PartitionedSourcesSpec.scala | 9 ++++++---
.../org/apache/pekko/kafka/scaladsl/ReconnectSpec.scala | 14 ++++++--------
.../org/apache/pekko/kafka/scaladsl/TransactionsSpec.scala | 9 ++++++---
4 files changed, 19 insertions(+), 15 deletions(-)
diff --git a/project/PekkoCoreDependency.scala
b/project/PekkoCoreDependency.scala
index fc61931d..b4a41697 100644
--- a/project/PekkoCoreDependency.scala
+++ b/project/PekkoCoreDependency.scala
@@ -20,5 +20,5 @@ import com.github.pjfanning.pekkobuild.PekkoDependency
object PekkoCoreDependency extends PekkoDependency {
override val checkProject: String = "pekko-cluster-sharding-typed"
override val module: Option[String] = None
- override val currentVersion: String = "2.0.0-M3"
+ override val currentVersion: String = "2.0.0-M4"
}
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 d9492b3f..0b2ed080 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
@@ -23,7 +23,7 @@ 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.KillSwitches
import pekko.stream.scaladsl.{ Keep, Sink, Source }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.stream.testkit.scaladsl.TestSink
@@ -389,7 +389,7 @@ class PartitionedSourcesSpec extends SpecBase with
TestcontainersKafkaLike with
awaitProduce(produce(topic, 1 to totalMessages))
val (queue, accumulator) = Source
- .queue[Long](8, OverflowStrategy.backpressure)
+ .queue[Long](totalMessages)
.toMat(Sink.fold(0)((c, _) => c + 1))(Keep.both)
.run()
@@ -401,7 +401,10 @@ class PartitionedSourcesSpec extends SpecBase with
TestcontainersKafkaLike with
case (tp, source) =>
source
.log(tp.toString, _.offset())
- .mapAsync(parallelism = 1)(rec =>
queue.offer(rec.offset()).map(_ => rec))
+ .map { rec =>
+ queue.offer(rec.offset())
+ rec
+ }
.map(_.value().toInt)
.takeWhile(_ < totalMessages, inclusive = true)
.map { value =>
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 9b0300fb..4909314e 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
@@ -15,11 +15,10 @@
package org.apache.pekko.kafka.scaladsl
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.scaladsl.{ Keep, Sink, Source, Tcp }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import pekko.stream.{ KillSwitches, OverflowStrategy, UniqueKillSwitch }
+import pekko.stream.{ BoundedSourceQueue, KillSwitches, UniqueKillSwitch }
import org.apache.kafka.clients.producer.ProducerRecord
import scala.concurrent.duration._
@@ -42,15 +41,14 @@ class ReconnectSpec extends SpecBase with
TestcontainersKafkaLike {
val firstBatch = 10
val messages = (1 to messagesProduced).map(_.toString)
// start a producer flow with a queue as source
- val producer: SourceQueueWithComplete[String] = Source
- .queue[String](messagesProduced, OverflowStrategy.backpressure)
+ val producer: BoundedSourceQueue[String] = Source
+ .queue[String](messagesProduced)
.map(msg => new ProducerRecord(topic1, partition0, DefaultKey, msg))
.to(Producer.plainSink(producerDefaults.withBootstrapServers(s"localhost:$proxyPort")))
.run()
- def offerInOrder(msgs: Seq[String]): Future[?] =
- if (msgs.isEmpty) Future.successful(Done)
- else producer.offer(msgs.head).flatMap(_ => offerInOrder(msgs.tail))
+ def offerInOrder(msgs: Seq[String]): Unit =
+ msgs.foreach(producer.offer)
// put one batch into the stream
offerInOrder(messages.take(firstBatch))
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 8a2b11eb..be8cadb1 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
@@ -22,7 +22,7 @@ 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.RestartSettings
import pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.consumer.ConsumerConfig
@@ -494,7 +494,7 @@ class TransactionsSpec extends SpecBase with
TestcontainersKafkaLike with Transa
val checkingGroup = createGroupId(2)
val (counterQueue, counterCompletion) = Source
- .queue[String](8, OverflowStrategy.fail)
+ .queue[String](totalMessages.toInt)
.scan(0L)((c, _) => c + 1)
.takeWhile(_ < totalMessages, inclusive = true)
.toMat(Sink.last)(Keep.both)
@@ -505,7 +505,10 @@ class TransactionsSpec extends SpecBase with
TestcontainersKafkaLike with Transa
.withGroupId(checkingGroup)
.withProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed"),
Subscriptions.topics(outTopic))
- .mapAsync(1)(el => counterQueue.offer(el.value()).map(_ => el))
+ .map { el =>
+ counterQueue.offer(el.value())
+ el
+ }
.scan(0L)((c, _) => c + 1)
.toMat(Sink.last)(DrainingControl.apply)
.run()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
