Replace unneeded throttle parameter with Source

Author: mdedetrich

backup-s3/src/test/scala/io/aiven/guardian/kafka/backup/s3/BackupClientSpec.scala

@@ -7,6 +7,7 @@ import akka.stream.alpakka.s3.S3Settings
 import akka.stream.alpakka.s3.scaladsl.S3
 import akka.stream.scaladsl.Keep
 import akka.stream.scaladsl.Sink
+import akka.stream.scaladsl.Source
 import akka.testkit.TestKitBase
 import com.softwaremill.diffx.generic.auto._
 import com.softwaremill.diffx.scalatest.DiffMatcher.matchTo
@@ -131,11 +132,11 @@ trait BackupClientSpec
     forAll(kafkaDataWithTimePeriodsGen(), s3ConfigGen(useVirtualDotHost, bucketPrefix)) {
       (kafkaDataWithTimePeriod: KafkaDataWithTimePeriod, s3Config: S3Config) =>
         logger.info(s"Data bucket is ${s3Config.dataBucket}")
-        val backupClient = new MockedS3BackupClientInterface(kafkaDataWithTimePeriod.data,
-                                                             kafkaDataWithTimePeriod.periodSlice,
-                                                             s3Config,
-                                                             Some(s3Settings),
-                                                             Some(_.throttle(ThrottleElements, ThrottleAmount))
+        val backupClient = new MockedS3BackupClientInterface(
+          Source(kafkaDataWithTimePeriod.data).throttle(ThrottleElements, ThrottleAmount),
+          kafkaDataWithTimePeriod.periodSlice,
+          s3Config,
+          Some(s3Settings)
         )
 
         val delay =

backup-s3/src/test/scala/io/aiven/guardian/kafka/backup/s3/MockedS3BackupClientInterface.scala

@@ -12,15 +12,12 @@ import io.aiven.guardian.kafka.s3.configs.{S3 => S3Config}
 import scala.concurrent.duration.FiniteDuration
 
 class MockedS3BackupClientInterface(
-    kafkaData: List[ReducedConsumerRecord],
+    kafkaData: Source[ReducedConsumerRecord, NotUsed],
     periodSlice: FiniteDuration,
     s3Config: S3Config,
-    maybeS3Settings: Option[S3Settings],
-    sourceTransform: Option[
-      Source[(ReducedConsumerRecord, Long), NotUsed] => Source[(ReducedConsumerRecord, Long), NotUsed]
-    ] = None
+    maybeS3Settings: Option[S3Settings]
 )(implicit val s3Headers: S3Headers)
-    extends BackupClient(maybeS3Settings)(new MockedKafkaClientInterface(kafkaData, sourceTransform),
+    extends BackupClient(maybeS3Settings)(new MockedKafkaClientInterface(kafkaData),
                                           Backup(periodSlice),
                                           s3Config,
                                           implicitly

core-backup/src/test/scala/io/aiven/guardian/kafka/backup/BackupClientInterfaceSpec.scala

@@ -40,7 +40,7 @@ class BackupClientInterfaceSpec
 
   property("Ordered Kafka events should produce at least one BackupStreamPosition.Boundary") {
     forAll(kafkaDataWithTimePeriodsGen()) { (kafkaDataWithTimePeriod: KafkaDataWithTimePeriod) =>
-      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(kafkaDataWithTimePeriod.data,
+      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(Source(kafkaDataWithTimePeriod.data),
                                                                     kafkaDataWithTimePeriod.periodSlice
       )
 
@@ -61,7 +61,7 @@ class BackupClientInterfaceSpec
     "Every ReducedConsumerRecord after a BackupStreamPosition.Boundary should be in the next consecutive time period"
   ) {
     forAll(kafkaDataWithTimePeriodsGen()) { (kafkaDataWithTimePeriod: KafkaDataWithTimePeriod) =>
-      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(kafkaDataWithTimePeriod.data,
+      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(Source(kafkaDataWithTimePeriod.data),
                                                                     kafkaDataWithTimePeriod.periodSlice
       )
 
@@ -95,7 +95,7 @@ class BackupClientInterfaceSpec
     "The time difference between two consecutive BackupStreamPosition.Middle's has to be less then the specified time period"
   ) {
     forAll(kafkaDataWithTimePeriodsGen()) { (kafkaDataWithTimePeriod: KafkaDataWithTimePeriod) =>
-      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(kafkaDataWithTimePeriod.data,
+      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(Source(kafkaDataWithTimePeriod.data),
                                                                     kafkaDataWithTimePeriod.periodSlice
       )
 
@@ -121,7 +121,7 @@ class BackupClientInterfaceSpec
 
   property("backup method completes flow correctly for all valid Kafka events") {
     forAll(kafkaDataWithTimePeriodsGen()) { (kafkaDataWithTimePeriod: KafkaDataWithTimePeriod) =>
-      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(kafkaDataWithTimePeriod.data,
+      val mock = new MockedBackupClientInterfaceWithMockedKafkaData(Source(kafkaDataWithTimePeriod.data),
                                                                     kafkaDataWithTimePeriod.periodSlice
       )
 

core-backup/src/test/scala/io/aiven/guardian/kafka/backup/MockedBackupClientInterface.scala

@@ -1,9 +1,11 @@
 package io.aiven.guardian.kafka.backup
 
 import akka.Done
+import akka.NotUsed
 import akka.actor.ActorSystem
 import akka.stream.scaladsl.Keep
 import akka.stream.scaladsl.Sink
+import akka.stream.scaladsl.Source
 import akka.util.ByteString
 import io.aiven.guardian.kafka.MockedKafkaClientInterface
 import io.aiven.guardian.kafka.backup.configs.Backup
@@ -78,6 +80,6 @@ class MockedBackupClientInterface(override val kafkaClientInterface: MockedKafka
 
 /** A `MockedBackupClientInterface` that also uses a mocked `KafkaClientInterface`
   */
-class MockedBackupClientInterfaceWithMockedKafkaData(kafkaData: List[ReducedConsumerRecord],
+class MockedBackupClientInterfaceWithMockedKafkaData(kafkaData: Source[ReducedConsumerRecord, NotUsed],
                                                      periodSlice: FiniteDuration
 ) extends MockedBackupClientInterface(new MockedKafkaClientInterface(kafkaData), periodSlice)

core/src/test/scala/io/aiven/guardian/kafka/MockedKafkaClientInterface.scala

@@ -19,12 +19,7 @@ import java.util.concurrent.ConcurrentLinkedQueue
   *   A function that allows you to transform the source in some way. Convenient for cases such as throttling. By
   *   default this is `None` so it just preserves the original source.
   */
-class MockedKafkaClientInterface(
-    kafkaData: List[ReducedConsumerRecord],
-    sourceTransform: Option[
-      Source[(ReducedConsumerRecord, Long), NotUsed] => Source[(ReducedConsumerRecord, Long), NotUsed]
-    ] = None
-) extends KafkaClientInterface {
+class MockedKafkaClientInterface(kafkaData: Source[ReducedConsumerRecord, NotUsed]) extends KafkaClientInterface {
 
   /** A collection that keeps track of whenever a cursor is committed
     */
@@ -42,17 +37,12 @@ class MockedKafkaClientInterface(
   /** @return
     *   A `SourceWithContext` that returns a Kafka Stream which automatically handles committing of cursors
     */
-  override def getSource: SourceWithContext[ReducedConsumerRecord, Long, Future[NotUsed]] = {
-    val source = Source(kafkaData.map { reducedConsumerRecord =>
-      (reducedConsumerRecord, reducedConsumerRecord.offset)
-    })
-
-    val finalSource = sourceTransform.fold(source)(block => block(source))
-
+  override def getSource: SourceWithContext[ReducedConsumerRecord, Long, Future[NotUsed]] =
     SourceWithContext
-      .fromTuples(finalSource)
+      .fromTuples(kafkaData.map { reducedConsumerRecord =>
+        (reducedConsumerRecord, reducedConsumerRecord.offset)
+      })
       .mapMaterializedValue(Future.successful)
-  }
 
   /** @return
     *   A `Sink` that allows you to commit a `CursorContext` to Kafka to signify you have processed a message