Update scalafmt-core to 3.8.0 in series/3.x (#1301)

* Update scalafmt-core to 3.8.0 in series/3.x

* Reformat with scalafmt 3.8.0

Executed command: scalafmt --non-interactive

* Add 'Reformat with scalafmt 3.8.0' to .git-blame-ignore-revs

.git-blame-ignore-revs

@@ -3,3 +3,6 @@ db0ea0ffd3d0956211a680941426f73ba7ec581b
 
 # Scala Steward: Reformat with scalafmt 3.7.17
 3dfecc9738dfc09b1bbd9a21b7f791f6ee7890b9
+
+# Scala Steward: Reformat with scalafmt 3.8.0
+5b4372fb8bddae0f2b4ce5e0f8b1b7e075899359

.scalafmt.conf

@@ -1,4 +1,4 @@
-version = "3.7.17"
+version = "3.8.0"
 
 # Scala 2 with -Xsource:3 compiler option
 runner.dialect = scala213source3

build.sbt

@@ -129,6 +129,7 @@ lazy val dependencySettings = Seq(
   },
   pomPostProcess := { (node: xml.Node) =>
     new xml.transform.RuleTransformer(new xml.transform.RewriteRule {
+
       def scopedDependency(e: xml.Elem): Boolean =
         e.label == "dependency" && e.child.exists(_.label == "scope")
 
@@ -137,6 +138,7 @@ lazy val dependencySettings = Seq(
           case e: xml.Elem if scopedDependency(e) => Nil
           case _                                  => Seq(node)
         }
+
     }).transform(node).head
   }
 )

docs/src/main/mdoc/consumers.md

@@ -80,8 +80,10 @@ If we have a Java Kafka deserializer, use `delegate` to create a `Deserializer`.
 ```scala mdoc:silent
 Deserializer.delegate[IO, String] {
   new KafkaDeserializer[String] {
+
     def deserialize(topic: String, data: Array[Byte]): String =
       new String(data)
+
   }
 }
 ```
@@ -92,10 +94,12 @@ If the deserializer performs _side effects_, follow with `suspend` to capture th
 Deserializer
   .delegate[IO, String] {
     new KafkaDeserializer[String] {
+
       def deserialize(topic: String, data: Array[Byte]): String = {
         println(s"deserializing record on topic $topic")
         new String(data)
       }
+
     }
   }
   .suspend
@@ -286,14 +290,14 @@ The recommended pattern for these use cases is by working on the `Chunk`s of rec
 
 ```scala mdoc:silent
 object ConsumerChunkExample extends IOApp.Simple {
+
   val run: IO[Unit] = {
     def processRecords(records: Chunk[ConsumerRecord[String, String]]): IO[CommitNow] =
       records.traverse(record => IO.println(s"Processing record: $record")).as(CommitNow)
 
-    KafkaConsumer.stream(consumerSettings)
-      .subscribeTo("topic")
-      .consumeChunk(processRecords)
+    KafkaConsumer.stream(consumerSettings).subscribeTo("topic").consumeChunk(processRecords)
   }
+
 }
 ```
 

docs/src/main/mdoc/producers.md

@@ -78,8 +78,10 @@ If we have a Java Kafka serializer, use `delegate` to create a `Serializer`.
 ```scala mdoc:silent
 Serializer.delegate[IO, String] {
   new KafkaSerializer[String] {
+
     def serialize(topic: String, value: String): Array[Byte] =
       value.getBytes("UTF-8")
+
   }
 }
 ```
@@ -90,10 +92,12 @@ If the serializer performs _side effects_, follow with `suspend` to capture them
 Serializer
   .delegate[IO, String] {
     new KafkaSerializer[String] {
+
       def serialize(topic: String, value: String): Array[Byte] = {
         println(s"serializing record on topic $topic")
         value.getBytes("UTF-8")
       }
+
     }
   }
   .suspend

docs/src/main/mdoc/quick-example.md

@@ -16,6 +16,7 @@ import fs2.kafka._
 import fs2.kafka.consumer.KafkaConsumeChunk.CommitNow
 
 object Main extends IOApp.Simple {
+
   val run: IO[Unit] = {
     val consumerSettings =
       ConsumerSettings[IO, String, String]
@@ -24,23 +25,28 @@ object Main extends IOApp.Simple {
         .withGroupId("group")
 
     val producerSettings =
-      ProducerSettings[IO, String, String]
-        .withBootstrapServers("localhost:9092")
+      ProducerSettings[IO, String, String].withBootstrapServers("localhost:9092")
 
-    def processRecords(producer: KafkaProducer[IO, String, String])(records: Chunk[ConsumerRecord[String, String]]): IO[CommitNow] = {
-      val producerRecords = records.map(consumerRecord => ProducerRecord("topic", consumerRecord.key, consumerRecord.value))
+    def processRecords(
+      producer: KafkaProducer[IO, String, String]
+    )(records: Chunk[ConsumerRecord[String, String]]): IO[CommitNow] = {
+      val producerRecords = records
+        .map(consumerRecord => ProducerRecord("topic", consumerRecord.key, consumerRecord.value))
       producer.produce(producerRecords).flatten.as(CommitNow)
     }
 
     val stream =
-      KafkaProducer.stream(producerSettings).evalMap { producer =>
-        KafkaConsumer
-          .stream(consumerSettings)
-          .subscribeTo("topic")
-          .consumeChunk(chunk => processRecords(producer)(chunk))
-      }
+      KafkaProducer
+        .stream(producerSettings)
+        .evalMap { producer =>
+          KafkaConsumer
+            .stream(consumerSettings)
+            .subscribeTo("topic")
+            .consumeChunk(chunk => processRecords(producer)(chunk))
+        }
 
     stream.compile.drain
   }
+
 }
 ```

modules/core/src/main/scala/fs2/kafka/CommittableOffset.scala

@@ -91,6 +91,7 @@ object CommittableOffset {
     val _commit            = commit
 
     new CommittableOffset[F] {
+
       override val topicPartition: TopicPartition =
         _topicPartition
 
@@ -119,6 +120,7 @@ object CommittableOffset {
           case None =>
             show"CommittableOffset($topicPartition -> $offsetAndMetadata)"
         }
+
     }
   }
 

modules/core/src/main/scala/fs2/kafka/CommittableOffsetBatch.scala

@@ -99,6 +99,7 @@ object CommittableOffsetBatch {
     val _commit                  = commit
 
     new CommittableOffsetBatch[F] {
+
       override def updated(that: CommittableOffset[F]): CommittableOffsetBatch[F] =
         CommittableOffsetBatch(
           _offsets.updated(that.topicPartition, that.offsetAndMetadata),
@@ -131,6 +132,7 @@ object CommittableOffsetBatch {
 
       override def toString: String =
         Show[CommittableOffsetBatch[F]].show(this)
+
     }
   }
 

modules/core/src/main/scala/fs2/kafka/Headers.scala

@@ -106,6 +106,7 @@ object Headers {
         toChain.iterator.toArray
 
       new KafkaHeaders {
+
         override def add(header: KafkaHeader): KafkaHeaders =
           throw new IllegalStateException("Headers#asJava is immutable")
 
@@ -128,6 +129,7 @@ object Headers {
 
         override def iterator(): java.util.Iterator[KafkaHeader] =
           array.iterator.asJava
+
       }
     }
 

modules/core/src/main/scala/fs2/kafka/Jitter.scala

@@ -36,8 +36,10 @@ object Jitter {
     F.delay(new java.util.Random())
       .map { random =>
         new Jitter[F] {
+
           override def withJitter(n: Double): F[Double] =
             F.delay(random.nextDouble()).map(_ * n)
+
         }
       }
 

modules/core/src/main/scala/fs2/kafka/KafkaProducerConnection.scala

@@ -120,6 +120,7 @@ object KafkaProducerConnection {
   ): Resource[F, KafkaProducerConnection[G]] =
     WithProducer(mk, settings).map { withProducer =>
       new KafkaProducerConnection[G] {
+
         override def produce[K, V](
           records: ProducerRecords[K, V]
         )(implicit
@@ -135,6 +136,7 @@ object KafkaProducerConnection {
 
         override def metrics: G[Map[MetricName, Metric]] =
           withProducer.blocking(_.metrics().asScala.toMap)
+
         override def withSerializers[K, V](
           keySerializer: KeySerializer[G, K],
           valueSerializer: ValueSerializer[G, V]
@@ -148,6 +150,7 @@ object KafkaProducerConnection {
 
         override def partitionsFor(topic: String): G[List[PartitionInfo]] =
           withProducer.blocking(_.partitionsFor(topic).asScala.toList)
+
       }
     }
 

modules/core/src/main/scala/fs2/kafka/TransactionalKafkaProducer.scala

@@ -98,6 +98,7 @@ object TransactionalKafkaProducer {
       WithTransactionalProducer(mk, settings)
     ).mapN { (keySerializer, valueSerializer, withProducer) =>
       new TransactionalKafkaProducer.WithoutOffsets[F, K, V] {
+
         override def produce(
           records: TransactionalProducerRecords[F, K, V]
         ): F[ProducerResult[K, V]] =
@@ -160,6 +161,7 @@ object TransactionalKafkaProducer {
 
         override def toString: String =
           "TransactionalKafkaProducer$" + System.identityHashCode(this)
+
       }
     }
 

modules/core/src/main/scala/fs2/kafka/internal/Logging.scala

@@ -24,6 +24,7 @@ private[kafka] object Logging {
     F.delay(LoggerFactory.getLogger(name))
       .map { logger =>
         new Logging[F] {
+
           override def log(entry: LogEntry): F[Unit] =
             F.delay {
               entry.level match {
@@ -41,6 +42,7 @@ private[kafka] object Logging {
                     logger.debug(entry.message)
               }
             }
+
         }
       }
 

modules/core/src/main/scala/fs2/kafka/internal/WithAdminClient.scala

@@ -29,8 +29,10 @@ private[kafka] object WithAdminClient {
       mk(settings).map { adminClient =>
         val withAdminClient =
           new WithAdminClient[G] {
+
             override def apply[A](f: AdminClient => KafkaFuture[A]): G[A] =
               G.delay(f(adminClient)).cancelable_
+
           }
 
         val close =

modules/core/src/main/scala/fs2/kafka/internal/WithConsumer.scala

@@ -31,8 +31,10 @@ private[kafka] object WithConsumer {
       Resource.make {
         mk(settings).map { consumer =>
           new WithConsumer[F] {
+
             override def blocking[A](f: KafkaByteConsumer => A): F[A] =
               b(f(consumer))
+
           }
         }
       }(_.blocking(_.close(settings.closeTimeout.toJava)))

modules/core/src/test/scala/fs2/kafka/KafkaAdminClientSpec.scala

@@ -86,12 +86,13 @@ final class KafkaAdminClientSpec extends BaseKafkaSpec {
                                                     .map((groupId, _))
                                                 }
               _ <- IO(assert(consumerGroupOffsetsPartitions.size == 1))
-              _ <- IO {
-                     adminClient
-                       .listConsumerGroupOffsets(consumerGroupId)
-                       .forPartitions(List(new TopicPartition("topic", 0)))
-                       .toString shouldBe "ListConsumerGroupOffsetsForPartitions(groupId = test-group-id, partitions = List(topic-0))"
-                   }
+              _ <-
+                IO {
+                  adminClient
+                    .listConsumerGroupOffsets(consumerGroupId)
+                    .forPartitions(List(new TopicPartition("topic", 0)))
+                    .toString shouldBe "ListConsumerGroupOffsetsForPartitions(groupId = test-group-id, partitions = List(topic-0))"
+                }
               partition0    = new TopicPartition(topic, 0)
               updatedOffset = new OffsetAndMetadata(0)
               _ <- adminClient

modules/core/src/test/scala/fs2/kafka/SerializerSpec.scala

@@ -72,10 +72,13 @@ final class SerializerSpec extends BaseCatsSpec {
       Serializer
         .delegate[IO, Int](
           new KafkaSerializer[Int] {
+
             override def close(): Unit                                                    = ()
             override def configure(props: java.util.Map[String, ?], isKey: Boolean): Unit = ()
+
             override def serialize(topic: String, int: Int): Array[Byte] =
               throw new RuntimeException
+
           }
         )
         .suspend
@@ -111,10 +114,13 @@ final class SerializerSpec extends BaseCatsSpec {
       Serializer
         .delegate[IO, Int](
           new KafkaSerializer[Int] {
+
             override def close(): Unit                                                    = ()
             override def configure(props: java.util.Map[String, ?], isKey: Boolean): Unit = ()
+
             override def serialize(topic: String, int: Int): Array[Byte] =
               throw new RuntimeException
+
           }
         )
         .suspend