format with new version

Author: novakov-alexey

.scalafmt.conf

@@ -1,6 +1,6 @@
 style       = defaultWithAlign
 maxColumn   = 120
-version = 3.8.3
+version = 3.8.4
 assumeStandardLibraryStripMargin = true
 align.stripMargin = true
 runner.dialect = scala3

modules/examples/src/main/scala/org/example/SocketTextStreamWordCount.scala

@@ -25,12 +25,11 @@ import org.apache.flink.configuration.ConfigConstants
 import org.apache.flink.configuration.RestOptions.BIND_PORT
 import scala.jdk.CollectionConverters.*
 
-/** This example shows an implementation of WordCount with data from a text
-  * socket. To run the example make sure that the service providing the text
-  * data is already up and running.
+/** This example shows an implementation of WordCount with data from a text socket. To run the example make sure that
+  * the service providing the text data is already up and running.
   *
-  * To start an example socket text stream on your local machine run netcat from
-  * a command line, where the parameter specifies the port number:
+  * To start an example socket text stream on your local machine run netcat from a command line, where the parameter
+  * specifies the port number:
   *
   * {{{
   *   nc -lk 9999

modules/examples/src/main/scala/org/example/TransactonIOs.scala

@@ -76,8 +76,8 @@ object TransactionsSource:
   def iterator: FromIteratorFunction[Transaction] =
     FromIteratorFunction[Transaction](
       (new Iterator[Transaction] with Serializable:
-        var rows = data.iterator
-        var timestamp = Timestamp.valueOf("2019-01-01 00:00:00").getTime
+        var rows       = data.iterator
+        var timestamp  = Timestamp.valueOf("2019-01-01 00:00:00").getTime
         val sixMinutes = 6.minutes.toMillis
 
         override def hasNext: Boolean = rows.hasNext

modules/examples/src/main/scala/org/example/connectedStreams.scala

@@ -19,16 +19,15 @@ import org.apache.flinkx.api.serializers.*
   val streamOfWords = env
     .addSource(TransactionsSource.iterator)
     .keyBy(_.accountId)
-  
+
   control
     .connect(streamOfWords)
     .flatMap(ControlFunction())
     .print()
 
   env.execute()
 
-class ControlFunction
-    extends RichCoFlatMapFunction[Transaction, Transaction, Transaction]:
+class ControlFunction extends RichCoFlatMapFunction[Transaction, Transaction, Transaction]:
 
   @transient lazy val state: ValueState[Double] = getRuntimeContext.getState(
     new ValueStateDescriptor(

modules/examples/src/main/scala/org/example/fileFilter.scala

@@ -28,7 +28,7 @@ class MyDefaultFileFilter extends Predicate[Path]:
 
 @main def filterFiles =
   val currentDirectory = File(".").getCanonicalPath
-  val inputBasePath = Path(s"$currentDirectory/input-table")
+  val inputBasePath    = Path(s"$currentDirectory/input-table")
   val fileSourceBuilder =
     FileSource.forRecordStreamFormat(
       TextLineInputFormat(),
@@ -37,9 +37,7 @@ class MyDefaultFileFilter extends Predicate[Path]:
 
   val fileSource = fileSourceBuilder
     .monitorContinuously(Duration.ofSeconds(2))
-    .setFileEnumerator(() =>
-      NonSplittingRecursiveEnumerator(MyDefaultFileFilter())
-    )
+    .setFileEnumerator(() => NonSplittingRecursiveEnumerator(MyDefaultFileFilter()))
     .build()
   val env =
     StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(Configuration())

modules/examples/src/main/scala/org/example/fraud/FraudDetectionJob.scala

@@ -18,7 +18,6 @@ package org.example.fraud
  * limitations under the License.
  */
 
-
 import java.io.File
 
 import org.apache.flinkx.api.*
@@ -73,7 +72,7 @@ import Givens.given
   conf.setString("execution.checkpointing.min-pause", "3s")
   conf.setString("state.backend", "filesystem")
 
-  val env = StreamExecutionEnvironment.getExecutionEnvironment //.createLocalEnvironmentWithWebUI(conf)
+  val env = StreamExecutionEnvironment.getExecutionEnvironment // .createLocalEnvironmentWithWebUI(conf)
 
   val transactions = env
     .addSource(TransactionsSource.iterator)
@@ -94,21 +93,20 @@ import Givens.given
 
 @main def fraudDetectionState() =
   val env = StreamExecutionEnvironment.getExecutionEnvironment
-  val savepoint = SavepointReader.read(env.getJavaEnv, "///tmp/savepoints/savepoint-827976-a94a8feb6c07",
-    HashMapStateBackend())
-  val keyedState = savepoint.readKeyedState("fraud-state", ReaderFunction(), TypeInformation.of(classOf[Long]), keyedStateInfo)
+  val savepoint =
+    SavepointReader.read(env.getJavaEnv, "///tmp/savepoints/savepoint-827976-a94a8feb6c07", HashMapStateBackend())
+  val keyedState =
+    savepoint.readKeyedState("fraud-state", ReaderFunction(), TypeInformation.of(classOf[Long]), keyedStateInfo)
   keyedState.print()
   env.execute()
 
 case class MaxTransaction(amount: Double, timestamp: Long)
 
-class MaxAggregate
-    extends AggregateFunction[Transaction, MaxTransaction, MaxTransaction]:
+class MaxAggregate extends AggregateFunction[Transaction, MaxTransaction, MaxTransaction]:
   override def createAccumulator(): MaxTransaction = MaxTransaction(0d, 0L)
 
   override def add(value: Transaction, accumulator: MaxTransaction): MaxTransaction =
-    if value.amount > accumulator._1 then
-      MaxTransaction(value.amount, value.timestamp)
+    if value.amount > accumulator._1 then MaxTransaction(value.amount, value.timestamp)
     else accumulator
 
   override def getResult(accumulator: MaxTransaction): MaxTransaction =

modules/examples/src/main/scala/org/example/fraud/FraudDetector.scala

@@ -55,8 +55,8 @@ case class FraudStateVars(
     timerState.clear()
 
 object FraudDetector:
-  val SmallAmount = 1.00
-  val LargeAmount = 500.00
+  val SmallAmount     = 1.00
+  val LargeAmount     = 500.00
   val OneMinute: Long = 1.minute.toMillis
 
   def readState(context: RuntimeContext): FraudStateVars =

modules/examples/src/main/scala/org/example/fraud/RunningAverage.scala

@@ -9,8 +9,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.configuration.Configuration
 import org.example.Transaction
 
-class RunningAverage
-    extends RichMapFunction[Transaction, (Transaction, Double)]:
+class RunningAverage extends RichMapFunction[Transaction, (Transaction, Double)]:
 
   given tranTypeInfo: TypeInformation[Transaction] =
     TypeInformation.of(classOf[Transaction])

modules/examples/src/main/scala/org/example/runningSum.scala

@@ -34,8 +34,7 @@ final case class TestEvent(
     bag: List[Int] = Nil
 )
 
-class CustomEventTimeTrigger[T, W <: TimeWindow](trigger: EventTimeTrigger)
-    extends Trigger[T, W]:
+class CustomEventTimeTrigger[T, W <: TimeWindow](trigger: EventTimeTrigger) extends Trigger[T, W]:
 
   override def onElement(
       element: T,
@@ -87,8 +86,7 @@ def windowAction(
   val output =
     reduced.copy(
       windowStart = window.getStart,
-      runningCount =
-        if reduced.runningCount > 0 then reduced.runningCount else 1
+      runningCount = if reduced.runningCount > 0 then reduced.runningCount else 1
     )
   println(
     s"\n{start: ${window.getStart} .. end: ${window.getEnd}, count: ${output.runningCount} \ninput: "
@@ -111,13 +109,11 @@ def reduceEvents(a: TestEvent, b: TestEvent) =
 @main def runningWindowedSum =
   val env = StreamExecutionEnvironment.getExecutionEnvironment
 
-  val windowSize = Time.of(10, TimeUnit.SECONDS)
+  val windowSize  = Time.of(10, TimeUnit.SECONDS)
   val windowSlide = Time.of(2, TimeUnit.SECONDS)
   val watermarkStrategy = WatermarkStrategy
     .forBoundedOutOfOrderness[TestEvent](Duration.ofSeconds(1000))
-    .withTimestampAssigner((event: TestEvent, streamRecordTimestamp: Long) =>
-      event.timestamp
-    )
+    .withTimestampAssigner((event: TestEvent, streamRecordTimestamp: Long) => event.timestamp)
 
   env
     .fromElements(
@@ -133,11 +129,10 @@ def reduceEvents(a: TestEvent, b: TestEvent) =
 
   env.execute()
 
-class RunningCountFunc(windowSize: Duration)
-    extends KeyedProcessFunction[Long, TestEvent, TestEvent]:
+class RunningCountFunc(windowSize: Duration) extends KeyedProcessFunction[Long, TestEvent, TestEvent]:
 
-  val oldEntriesCleanupInterval = 1000L
-  var minTimestamp: ValueState[Long] = _
+  val oldEntriesCleanupInterval         = 1000L
+  var minTimestamp: ValueState[Long]    = _
   var timeToCount: MapState[Long, Long] = _
   override def open(parameters: Configuration): Unit =
     timeToCount = getRuntimeContext.getMapState(
@@ -158,8 +153,7 @@ class RunningCountFunc(windowSize: Duration)
       out: Collector[TestEvent]
   ): Unit =
     val currentCount =
-      if timeToCount.contains(event.timestamp) then
-        timeToCount.get(event.timestamp)
+      if timeToCount.contains(event.timestamp) then timeToCount.get(event.timestamp)
       else 0
     timeToCount.put(event.timestamp, currentCount + 1)
 

modules/examples/src/main/scala/org/example/troubleshooting/fakeKafkaSource.scala

@@ -37,14 +37,14 @@ class FakeKafkaSource(
     (0 to FakeKafkaSource.NO_OF_PARTITIONS).filter(
       _ % numberOfParallelSubtasks == indexOfThisSubtask
     )
-  
+
   val rand = Random(seed)
 
   @transient @volatile var cancelled = false
 
   override def open(parameters: Configuration): Unit =
     println(s"Now reading from partitions: $assignedPartitions")
-    
+
   override def run(ctx: SourceContext[FakeKafkaRecord]): Unit =
     if assignedPartitions.nonEmpty then
       while !cancelled do {
@@ -61,8 +61,7 @@ class FakeKafkaSource(
           var serializedMeasurement =
             serializedMeasurements(rand.nextInt(serializedMeasurements.length))
 
-          if rand.nextFloat() > 1 - poisonPillRate then
-            serializedMeasurement = Arrays.copyOf(serializedMeasurement, 10)
+          if rand.nextFloat() > 1 - poisonPillRate then serializedMeasurement = Arrays.copyOf(serializedMeasurement, 10)
 
           (ctx.getCheckpointLock()).synchronized {
             ctx.collect(