You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/01/17 21:33:09 UTC
(pekko-connectors-kafka) 02/03: Reformat with scalafmt 3.10.4
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
commit f6aff22f025ba7779c93785bfdea850cd3940db3
Author: Scala Steward <sc...@virtuslab.com>
AuthorDate: Sat Jan 17 16:23:13 2026 +0000
Reformat with scalafmt 3.10.4
Executed command: scalafmt --non-interactive
---
.../benchmarks/KafkaTransactionFixtureGen.scala | 5 +---
.../PekkoConnectorsCommittableSinkFixtures.scala | 5 +---
.../ReactiveKafkaTransactionFixtures.scala | 5 +---
.../kafka/internal/CommitObservationLogic.scala | 3 +-
.../kafka/internal/ConsumerProgressTracking.scala | 33 +++++++++++-----------
.../pekko/kafka/internal/KafkaConsumerActor.scala | 9 +-----
.../pekko/kafka/internal/MessageBuilder.scala | 6 +---
.../pekko/kafka/scaladsl/MetadataClient.scala | 32 ++++++++++-----------
.../pekko/kafka/scaladsl/Transactional.scala | 5 +---
.../kafka/TransactionsPartitionedSourceSpec.scala | 10 +++----
.../pekko/kafka/TransactionsSourceSpec.scala | 13 ++++-----
project/ParadoxSettings.scala | 15 +++++-----
project/ProjectSettings.scala | 3 +-
.../testkit/internal/KafkaTestKitChecks.scala | 5 +---
.../test/scala/docs/scaladsl/FetchMetadata.scala | 4 +--
.../org/apache/pekko/kafka/TransactionsOps.scala | 6 ++--
.../kafka/internal/CommitCollectorStageSpec.scala | 15 ++++++----
.../kafka/scaladsl/ConnectionCheckerSpec.scala | 3 +-
.../pekko/kafka/scaladsl/RebalanceExtSpec.scala | 3 +-
.../pekko/kafka/scaladsl/TransactionsSpec.scala | 21 +++++++-------
20 files changed, 91 insertions(+), 110 deletions(-)
diff --git a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.scala b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.scala
index 52451061..51d439bf 100644
--- a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.scala
+++ b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.scala
@@ -22,10 +22,7 @@ import org.apache.kafka.clients.consumer.{ ConsumerConfig, KafkaConsumer }
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerConfig }
import org.apache.kafka.common.IsolationLevel
import org.apache.kafka.common.serialization.{
- ByteArrayDeserializer,
- ByteArraySerializer,
- StringDeserializer,
- StringSerializer
+ ByteArrayDeserializer, ByteArraySerializer, StringDeserializer, StringSerializer
}
import scala.jdk.CollectionConverters._
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 46074109..35dfb8bc 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
@@ -30,10 +30,7 @@ import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.{
- ByteArrayDeserializer,
- ByteArraySerializer,
- StringDeserializer,
- StringSerializer
+ ByteArrayDeserializer, ByteArraySerializer, StringDeserializer, StringSerializer
}
import scala.concurrent.duration._
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 4b5be6b6..3cdf2301 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
@@ -26,10 +26,7 @@ import pekko.kafka.{ ConsumerMessage, ConsumerSettings, ProducerSettings, Subscr
import pekko.stream.scaladsl.{ Flow, Source }
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.{
- ByteArrayDeserializer,
- ByteArraySerializer,
- StringDeserializer,
- StringSerializer
+ ByteArrayDeserializer, ByteArraySerializer, StringDeserializer, StringSerializer
}
import scala.concurrent.duration.FiniteDuration
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 e0293427..a1e2a175 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
@@ -69,7 +69,8 @@ private[internal] trait CommitObservationLogic { self: GraphStageLogic =>
deferredOffsets = deferredOffsets + (gtp -> committable)
offsetBatch = offsetBatch.updated(dOffset)
case Some(dOffsetBatch: CommittableOffsetBatch)
- if dOffsetBatch.offsets.contains(gtp) && dOffsetBatch.offsets
+ if dOffsetBatch.offsets.contains(gtp) &&
+ dOffsetBatch.offsets
.get(gtp)
.head < offset =>
deferredOffsets = deferredOffsets + (gtp -> committable)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerProgressTracking.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerProgressTracking.scala
index 7ca6a03e..591ef3fc 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerProgressTracking.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerProgressTracking.scala
@@ -88,22 +88,23 @@ final class ConsumerProgressTrackerImpl extends ConsumerProgressTracking {
}
override def received[K, V](received: ConsumerRecords[K, V]): Unit = {
- receivedMessagesImpl = receivedMessagesImpl ++ received
- .partitions()
- .asScala
- // only tracks the partitions that are currently assigned, as assignment is a synchronous interaction and polls
- // for an old consumer group epoch will not return (we get to make polls for the current generation). Supposing a
- // revoke completes and then the poll() is received for a previous epoch, we drop the records here (partitions
- // are no longer assigned to the consumer). If instead we get a poll() and then a revoke, we only track the
- // offsets for that short period of time and then they are revoked, so that is also safe.
- .intersect(assignedPartitions)
- .map(tp => (tp, received.records(tp)))
- // get the last record, its the largest offset/most recent timestamp
- .map { case (partition, records) => (partition, records.get(records.size() - 1)) }
- .map {
- case (partition, record) =>
- partition -> SafeOffsetAndTimestamp(record.offset(), record.timestamp())
- }
+ receivedMessagesImpl = receivedMessagesImpl ++
+ received
+ .partitions()
+ .asScala
+ // only tracks the partitions that are currently assigned, as assignment is a synchronous interaction and polls
+ // for an old consumer group epoch will not return (we get to make polls for the current generation). Supposing a
+ // revoke completes and then the poll() is received for a previous epoch, we drop the records here (partitions
+ // are no longer assigned to the consumer). If instead we get a poll() and then a revoke, we only track the
+ // offsets for that short period of time and then they are revoked, so that is also safe.
+ .intersect(assignedPartitions)
+ .map(tp => (tp, received.records(tp)))
+ // get the last record, its the largest offset/most recent timestamp
+ .map { case (partition, records) => (partition, records.get(records.size() - 1)) }
+ .map {
+ case (partition, record) =>
+ partition -> SafeOffsetAndTimestamp(record.offset(), record.timestamp())
+ }
}
override def commitRequested(offsets: Map[TopicPartition, OffsetAndMetadata]): Unit = {
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 fa1356a5..b15f6b3d 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
@@ -21,14 +21,7 @@ import org.apache.pekko
import pekko.Done
import pekko.actor.Status.Failure
import pekko.actor.{
- Actor,
- ActorRef,
- DeadLetterSuppression,
- NoSerializationVerificationNeeded,
- Stash,
- Status,
- Terminated,
- Timers
+ Actor, ActorRef, DeadLetterSuppression, NoSerializationVerificationNeeded, Stash, Status, Terminated, Timers
}
import pekko.annotation.InternalApi
import pekko.event.LoggingReceive
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 f985faac..89273b89 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
@@ -21,11 +21,7 @@ import pekko.Done
import pekko.annotation.InternalApi
import pekko.kafka.ConsumerMessage
import pekko.kafka.ConsumerMessage.{
- CommittableMessage,
- CommittableOffsetMetadata,
- GroupTopicPartition,
- TransactionalMessage,
- _
+ CommittableMessage, CommittableOffsetMetadata, GroupTopicPartition, TransactionalMessage, _
}
import org.apache.kafka.clients.consumer.{ ConsumerRecord, OffsetAndMetadata }
import org.apache.kafka.common.TopicPartition
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 bb3c049c..a3e18b96 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
@@ -57,23 +57,21 @@ class MetadataClient private (consumerActor: ActorRef, timeout: Timeout, managed
getEndOffsets(Set(partition))
.map(endOffsets => endOffsets(partition))
- def listTopics(): Future[Map[String, List[PartitionInfo]]] =
- (consumerActor ? ListTopics)(timeout)
- .mapTo[Topics]
- .map(_.response)
- .flatMap {
- case Success(res) => Future.successful(res)
- case Failure(e) => Future.failed(e)
- }(ExecutionContext.parasitic)
-
- def getPartitionsFor(topic: String): Future[List[PartitionInfo]] =
- (consumerActor ? GetPartitionsFor(topic))(timeout)
- .mapTo[PartitionsFor]
- .map(_.response)
- .flatMap {
- case Success(res) => Future.successful(res)
- case Failure(e) => Future.failed(e)
- }(ExecutionContext.parasitic)
+ def listTopics(): Future[Map[String, List[PartitionInfo]]] = (consumerActor ? ListTopics)(timeout)
+ .mapTo[Topics]
+ .map(_.response)
+ .flatMap {
+ case Success(res) => Future.successful(res)
+ case Failure(e) => Future.failed(e)
+ }(ExecutionContext.parasitic)
+
+ def getPartitionsFor(topic: String): Future[List[PartitionInfo]] = (consumerActor ? GetPartitionsFor(topic))(timeout)
+ .mapTo[PartitionsFor]
+ .map(_.response)
+ .flatMap {
+ case Success(res) => Future.successful(res)
+ case Failure(e) => Future.failed(e)
+ }(ExecutionContext.parasitic)
def getCommittedOffsets(partitions: Set[TopicPartition]): Future[Map[TopicPartition, OffsetAndMetadata]] =
(consumerActor ? GetCommittedOffsets(partitions))(timeout)
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 22039eee..752ee6a2 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
@@ -19,10 +19,7 @@ import pekko.annotation.{ ApiMayChange, InternalApi }
import pekko.kafka.ConsumerMessage.{ PartitionOffset, TransactionalMessage }
import pekko.kafka.ProducerMessage._
import pekko.kafka.internal.{
- TransactionalProducerStage,
- TransactionalSource,
- TransactionalSourceWithOffsetContext,
- TransactionalSubSource
+ TransactionalProducerStage, TransactionalSource, TransactionalSourceWithOffsetContext, TransactionalSubSource
}
import pekko.kafka.scaladsl.Consumer.Control
import pekko.kafka.{ AutoSubscription, ConsumerMessage, ConsumerSettings, ProducerSettings, Subscription }
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 e6081dce..22d6d35b 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
@@ -54,7 +54,8 @@ class TransactionsPartitionedSourceSpec
.withInternalTopicsReplicationFactor(replicationFactor)
"A multi-broker consume-transform-produce cycle" must {
- "provide consistency when multiple partitioned transactional streams are being restarted" in assertAllStagesStopped {
+ "provide consistency when multiple partitioned transactional streams are being restarted" in
+ assertAllStagesStopped {
// It's possible to get into a livelock situation where the `restartAfter` interval causes transactions to abort
// over and over. This can happen when there are a few partitions left to process and they can never be fully
// processed because we always restart the stream before the transaction can be completed successfully.
@@ -74,10 +75,9 @@ 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 =>
- produce(sourceTopic, range = 1 to elements, partition = part)
- }
+ val producers: immutable.Seq[Future[Done]] = (0 until sourcePartitions).map { part =>
+ produce(sourceTopic, range = 1 to elements, partition = part)
+ }
Await.result(Future.sequence(producers), 4.minute)
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 552ace46..d8c02fdf 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
@@ -72,13 +72,12 @@ class TransactionsSourceSpec
val restartAfter = 10 * 1000
val partitionSize = elements / sourcePartitions
- val producers: immutable.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")
- produce(sourceTopic, rangeStart to rangeEnd, part)
- }
+ val producers: immutable.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")
+ produce(sourceTopic, rangeStart to rangeEnd, part)
+ }
Await.result(Future.sequence(producers), 1.minute)
diff --git a/project/ParadoxSettings.scala b/project/ParadoxSettings.scala
index 08ed511b..ee75aa33 100644
--- a/project/ParadoxSettings.scala
+++ b/project/ParadoxSettings.scala
@@ -10,11 +10,7 @@
import Versions._
import com.lightbend.paradox.apidoc.ApidocPlugin.autoImport.apidocRootPackage
import com.lightbend.paradox.sbt.ParadoxPlugin.autoImport.{
- paradox,
- paradoxGroups,
- paradoxMarkdownToHtml,
- paradoxProperties,
- paradoxRoots
+ paradox, paradoxGroups, paradoxMarkdownToHtml, paradoxProperties, paradoxRoots
}
import org.apache.pekko.PekkoParadoxPlugin.autoImport._
import sbt._
@@ -38,8 +34,10 @@ object ParadoxSettings {
"extref.pekko.base_url" -> s"$pekkoDocs/pekko/$pekkoVersionForDocs/%s",
"scaladoc.org.apache.pekko.base_url" -> s"$pekkoAPI/pekko/$pekkoVersionForDocs/",
"javadoc.org.apache.pekko.base_url" -> s"$pekkoAPI/pekko/$pekkoVersionForDocs/",
- "scaladoc.org.apache.pekko.kafka.base_url" -> s"$pekkoAPI/pekko-connectors-kafka/$pekkoConnectorsKafkaVersionForDocs/",
- "javadoc.org.apache.pekko.kafka.base_url" -> s"$pekkoAPI/pekko-connectors-kafka/$pekkoConnectorsKafkaVersionForDocs/",
+ "scaladoc.org.apache.pekko.kafka.base_url" ->
+ s"$pekkoAPI/pekko-connectors-kafka/$pekkoConnectorsKafkaVersionForDocs/",
+ "javadoc.org.apache.pekko.kafka.base_url" ->
+ s"$pekkoAPI/pekko-connectors-kafka/$pekkoConnectorsKafkaVersionForDocs/",
"javadoc.org.apache.pekko.link_style" -> "direct",
"extref.pekko-management.base_url" -> s"$pekkoDocs/pekko-management/$pekkoManagementVersionForDocs/%s",
// Kafka
@@ -56,7 +54,8 @@ object ParadoxSettings {
"scaladoc.com.typesafe.config.base_url" -> s"https://lightbend.github.io/config/latest/api/",
// Testcontainers
"testcontainers.version" -> testcontainersVersion,
- "javadoc.org.testcontainers.containers.base_url" -> s"https://www.javadoc.io/doc/org.testcontainers/testcontainers/$testcontainersVersion/",
+ "javadoc.org.testcontainers.containers.base_url" ->
+ s"https://www.javadoc.io/doc/org.testcontainers/testcontainers/$testcontainersVersion/",
"javadoc.org.testcontainers.containers.link_style" -> "direct"))
val sourceGeneratorSettings = Seq(
diff --git a/project/ProjectSettings.scala b/project/ProjectSettings.scala
index ff0fe5ff..ec98d672 100644
--- a/project/ProjectSettings.scala
+++ b/project/ProjectSettings.scala
@@ -76,7 +76,8 @@ object ProjectSettings extends AutoPlugin {
"dev@pekko.apache.org",
url("https://github.com/apache/pekko-connectors-kafka/graphs/contributors")),
startYear := Some(2022),
- description := "Apache Pekko Kafka Connector is a Reactive Enterprise Integration library for Java and Scala, based on Reactive Streams and Apache Pekko.",
+ description :=
+ "Apache Pekko Kafka Connector is a Reactive Enterprise Integration library for Java and Scala, based on Reactive Streams and Apache Pekko.",
crossScalaVersions := Seq(Scala213, Scala3),
scalaVersion := Scala213,
crossVersion := CrossVersion.binary,
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
index 93478018..f34e9aaf 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
@@ -18,10 +18,7 @@ import java.util.Collections
import java.util.concurrent.TimeUnit
import org.apache.kafka.clients.admin.{
- Admin,
- ConsumerGroupDescription,
- DescribeClusterResult,
- DescribeConsumerGroupsOptions
+ Admin, ConsumerGroupDescription, DescribeClusterResult, DescribeConsumerGroupsOptions
}
import org.slf4j.Logger
diff --git a/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala b/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
index 55e9f8b7..436a3cce 100644
--- a/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
+++ b/tests/src/test/scala/docs/scaladsl/FetchMetadata.scala
@@ -120,8 +120,8 @@ class FetchMetadata extends DocsSpecBase with TestcontainersKafkaLike with TryVa
val partition = 0
val tp = new TopicPartition(topic, partition)
- val topicsFuture: Future[Metadata.EndOffsets] =
- (consumer ? Metadata.GetEndOffsets(Set(tp))).mapTo[Metadata.EndOffsets]
+ val topicsFuture
+ : Future[Metadata.EndOffsets] = (consumer ? Metadata.GetEndOffsets(Set(tp))).mapTo[Metadata.EndOffsets]
val response = topicsFuture.futureValue.response
(response should be).a(Symbol("success"))
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 d2c35fd4..8d4ec3c6 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/TransactionsOps.scala
@@ -93,9 +93,9 @@ trait TransactionsOps extends TestSuite with Matchers {
.idleTimeout(idleTimeout)
.map { msg =>
ProducerMessage.single(new ProducerRecord[String, String](sinkTopic,
- msg.record.partition(),
- msg.record.key(),
- msg.record.value),
+ msg.record.partition(),
+ msg.record.key(),
+ msg.record.value),
msg.partitionOffset)
}
.via(Transactional.flow(producerSettings, transactionalId))
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 ec7fc6fd..bca4f25a 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
@@ -190,7 +190,8 @@ class CommitCollectorStageSpec(_system: ActorSystem)
control.shutdown().futureValue shouldBe Done
}
- "batch commit all buffered elements if upstream has suddenly completed with delayed commits" in assertAllStagesStopped {
+ "batch commit all buffered elements if upstream has suddenly completed with delayed commits" in
+ assertAllStagesStopped {
val (sourceProbe, control, sinkProbe) = streamProbes(settings)
val committer = new TestBatchCommitter(settings, () => 50.millis)
@@ -285,7 +286,8 @@ class CommitCollectorStageSpec(_system: ActorSystem)
// downstream out of order
val lastBatch = batches.maxBy(_.offsets.values.last)
- (lastBatch.offsets.values.last shouldBe batch2
+ (lastBatch.offsets.values.last shouldBe
+ batch2
.asInstanceOf[CommittableOffsetBatch]
.offsets
.head
@@ -294,7 +296,8 @@ class CommitCollectorStageSpec(_system: ActorSystem)
control.shutdown().futureValue shouldBe Done
}
- "only commit when the next offset is observed in a CommittableOffset preceded by a CommittableOffsetBatch" in assertAllStagesStopped {
+ "only commit when the next offset is observed in a CommittableOffset preceded by a CommittableOffsetBatch" in
+ assertAllStagesStopped {
val (sourceProbe, control, sinkProbe, offsetFactory) = streamProbesWithOffsetFactory(settings)
// create a mix of single offsets and batches of 1
val (batch1, msg2, batch3) =
@@ -320,7 +323,8 @@ class CommitCollectorStageSpec(_system: ActorSystem)
control.shutdown().futureValue shouldBe Done
}
- "only commit when the next offset is observed in a CommittableOffsetBatch preceded by a CommittableOffset" in assertAllStagesStopped {
+ "only commit when the next offset is observed in a CommittableOffsetBatch preceded by a CommittableOffset" in
+ assertAllStagesStopped {
val (sourceProbe, control, sinkProbe, offsetFactory) = streamProbesWithOffsetFactory(settings)
// create a mix of single offsets and batches of 1
val (msg1, batch2, msg3) =
@@ -340,7 +344,8 @@ class CommitCollectorStageSpec(_system: ActorSystem)
// downstream out of order
val lastBatch = batches.maxBy(_.offsets.values.last)
- (lastBatch.offsets.values.last shouldBe batch2
+ (lastBatch.offsets.values.last shouldBe
+ batch2
.asInstanceOf[CommittableOffsetBatch]
.offsets
.head
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 4725d418..217a7e21 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
@@ -74,7 +74,8 @@ class ConnectionCheckerSpec extends SpecBase with TestcontainersKafkaPerClassLik
Await.ready(control.isShutdown.zip(futDone), failingDetectionTime)
}
- "fail stream and control.isShutdown when kafka down and not recover during max retries exceeded" in assertAllStagesStopped {
+ "fail stream and control.isShutdown when kafka down and not recover during max retries exceeded" in
+ assertAllStagesStopped {
startCluster()
val msg = "hello"
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 231aed24..8c4f6fbd 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
@@ -198,7 +198,8 @@ class RebalanceExtSpec extends SpecBase with TestcontainersKafkaLike with Inside
"Fetched records" must {
- "no messages should be lost when two consumers consume from one topic and two partitions and one consumer aborts mid-stream" in assertAllStagesStopped {
+ "no messages should be lost when two consumers consume from one topic and two partitions and one consumer aborts mid-stream" in
+ assertAllStagesStopped {
val topicCount = 1
val partitionCount = 2
val perPartitionMessageCount = 9
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 a1d52fec..8a2b11eb 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
@@ -150,7 +150,8 @@ class TransactionsSpec extends SpecBase with TestcontainersKafkaLike with Transa
}
}
- "complete with messages filtered out and transient failure causing an abort with restartable source" in assertAllStagesStopped {
+ "complete with messages filtered out and transient failure causing an abort with restartable source" in
+ assertAllStagesStopped {
val sourceTopic = createTopic(1)
val sinkTopic = createTopic(2)
val group = createGroupId(1)
@@ -314,9 +315,9 @@ class TransactionsSpec extends SpecBase with TestcontainersKafkaLike with Transa
source
.map { msg =>
ProducerMessage.single(new ProducerRecord[String, String](sinkTopic,
- msg.record.partition(),
- msg.record.key(),
- msg.record.value),
+ msg.record.partition(),
+ msg.record.key(),
+ msg.record.value),
msg.partitionOffset)
}
.runWith(Transactional.sink(producerDefaults, transactionalId))
@@ -380,9 +381,9 @@ class TransactionsSpec extends SpecBase with TestcontainersKafkaLike with Transa
throw new Exception("sub source failure")
}
ProducerMessage.single(new ProducerRecord[String, String](sinkTopic,
- msg.record.partition(),
- msg.record.key(),
- msg.record.value),
+ msg.record.partition(),
+ msg.record.key(),
+ msg.record.value),
msg.partitionOffset)
}
.runWith(Transactional.sink(producerDefaults, transactionalId))
@@ -453,9 +454,9 @@ class TransactionsSpec extends SpecBase with TestcontainersKafkaLike with Transa
source
.map { msg =>
ProducerMessage.single(new ProducerRecord[String, String](outTopic,
- msg.record.partition(),
- msg.record.key(),
- msg.record.value() + "-out"),
+ msg.record.partition(),
+ msg.record.key(),
+ msg.record.value() + "-out"),
msg.partitionOffset)
}
.to(Transactional.sink(producerDefaults, transactionalId))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org