Revert "Avoid filtering by lastSeenFile where a post process action is configured (#192)" (#210)

This reverts commit 771ed92

Author: davidsloan

kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/source/files/CloudSourceFileQueue.scala

@@ -14,14 +14,14 @@
  * limitations under the License.
  */
 package io.lenses.streamreactor.connect.cloud.common.source.files
+
 import cats.implicits.catsSyntaxEitherId
 import cats.implicits.catsSyntaxOptionId
 import cats.implicits.toShow
 import com.typesafe.scalalogging.LazyLogging
 import io.lenses.streamreactor.connect.cloud.common.config.ConnectorTaskId
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocation
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocationValidator
-import io.lenses.streamreactor.connect.cloud.common.source.config.PostProcessAction
 import io.lenses.streamreactor.connect.cloud.common.storage.FileListError
 import io.lenses.streamreactor.connect.cloud.common.storage.FileMetadata
 import io.lenses.streamreactor.connect.cloud.common.storage.ListOfKeysResponse
@@ -32,14 +32,6 @@ trait SourceFileQueue {
   def next(): Either[FileListError, Option[CloudLocation]]
 }
 
-/**
-  * A tracker for the last seen file.
-  *
-  * @param lastSeenFile An optional metadata object representing the last seen file.
-  * @tparam SM The type of the file metadata.
-  */
-case class LastSeenFileTracker[SM <: FileMetadata](var lastSeenFile: Option[SM])
-
 /**
   * Blocking processor for queues of operations.  Used to ensure consistency.
   * Will block any further writes by the current file until the remote has caught up.
@@ -57,7 +49,7 @@ class CloudSourceFileQueue[SM <: FileMetadata] private (
     *                            files will be cleaned up after processing, removing the need to track the last seen file.
     *                            In such cases, this parameter will be None.
     */
-  private var lastSeenFileTracker: Option[LastSeenFileTracker[SM]],
+  private var lastSeenFile: Option[SM],
 )(
   implicit
   cloudLocationValidator: CloudLocationValidator,
@@ -70,7 +62,7 @@ class CloudSourceFileQueue[SM <: FileMetadata] private (
   )(
     implicit
     cloudLocationValidator: CloudLocationValidator,
-  ) = this(taskId, batchListerFn, Seq.empty, Some(LastSeenFileTracker[SM](None)))
+  ) = this(taskId, batchListerFn, Seq.empty, None)
 
   override def next(): Either[FileListError, Option[CloudLocation]] =
     files match {
@@ -90,11 +82,10 @@ class CloudSourceFileQueue[SM <: FileMetadata] private (
 
   private def retrieveNextFile(
   ): Either[FileListError, Option[CloudLocation]] = {
-    val nextBatch: Either[FileListError, Option[ListResponse[String, SM]]] =
-      batchListerFn(lastSeenFileTracker.flatMap(_.lastSeenFile))
+    val nextBatch: Either[FileListError, Option[ListResponse[String, SM]]] = batchListerFn(lastSeenFile)
     nextBatch.flatMap {
       case Some(ListOfKeysResponse(bucket, prefix, value, meta)) =>
-        lastSeenFileTracker = lastSeenFileTracker.map(_.copy(lastSeenFile = meta.some))
+        lastSeenFile = meta.some
         files = value.map(path =>
           CloudLocation(
             bucket,
@@ -116,11 +107,10 @@ class CloudSourceFileQueue[SM <: FileMetadata] private (
 
 object CloudSourceFileQueue {
   def from[SM <: FileMetadata](
-    batchListerFn:          Option[SM] => Either[FileListError, Option[ListOfKeysResponse[SM]]],
-    storageInterface:       StorageInterface[SM],
-    startingFile:           CloudLocation,
-    taskId:                 ConnectorTaskId,
-    maybePostProcessAction: Option[PostProcessAction],
+    batchListerFn:    Option[SM] => Either[FileListError, Option[ListOfKeysResponse[SM]]],
+    storageInterface: StorageInterface[SM],
+    startingFile:     CloudLocation,
+    taskId:           ConnectorTaskId,
   )(
     implicit
     cloudLocationValidator: CloudLocationValidator,
@@ -133,15 +123,7 @@ object CloudSourceFileQueue {
       case _ =>
         Option.empty[SM]
     }
-    new CloudSourceFileQueue[SM](
-      taskId,
-      batchListerFn,
-      Seq(startingFile),
-      // Creates an instance of LastSeenFileTracker if no PostProcessAction is set
-      // If PostProcessAction is set then files will be cleaned up after processing,
-      // which removes the requirement to seek through to the last seen file.
-      Option.when(maybePostProcessAction.isEmpty)(LastSeenFileTracker[SM](lastSeen)),
-    )
+    new CloudSourceFileQueue(taskId, batchListerFn, Seq(startingFile), lastSeen)
   }
 
 }

kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/source/state/ReaderManagerBuilder.scala

@@ -72,7 +72,6 @@ object ReaderManagerBuilder extends LazyLogging {
           storageInterface,
           location,
           connectorTaskId,
-          adaptedSbo.postProcessAction,
         )
       }
       sourceFileQueue <- IO.fromEither(

kafka-connect-cloud-common/src/test/scala/io/lenses/streamreactor/connect/cloud/common/source/files/CloudSourceFileQueueTest.scala

@@ -22,13 +22,11 @@ import io.lenses.streamreactor.connect.cloud.common.config.ConnectorTaskId
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocation
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocationValidator
 import io.lenses.streamreactor.connect.cloud.common.sink.seek.TestFileMetadata
-import io.lenses.streamreactor.connect.cloud.common.source.config.DeletePostProcessAction
 import io.lenses.streamreactor.connect.cloud.common.storage.FileListError
 import io.lenses.streamreactor.connect.cloud.common.storage.ListOfKeysResponse
 import io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface
 import io.lenses.streamreactor.connect.cloud.common.utils.SampleData
 import org.mockito.ArgumentMatchers._
-import org.mockito.InOrder
 import org.mockito.MockitoSugar
 import org.scalatest.BeforeAndAfter
 import org.scalatest.flatspec.AnyFlatSpec
@@ -85,7 +83,12 @@ class CloudSourceFileQueueTest extends AnyFlatSpec with Matchers with MockitoSug
       any[Option[TestFileMetadata]],
     )
     // file 0 = 0.json
-    verifyRefreshAndRollover(batchListerFn, sourceFileQueue, order)
+    sourceFileQueue.next() should be(Right(Some(fileLocs(0).atLine(-1).withTimestamp(lastModified))))
+    order.verify(batchListerFn)(none)
+
+    // file 1 = 1.json
+    sourceFileQueue.next() should be(Right(Some(fileLocs(1).atLine(-1).withTimestamp(lastModified))))
+    order.verifyNoMoreInteractions()
 
     // file 2 = 2.json
     sourceFileQueue.next() should be(Right(Some(fileLocs(2).atLine(-1).withTimestamp(lastModified))))
@@ -124,7 +127,6 @@ class CloudSourceFileQueueTest extends AnyFlatSpec with Matchers with MockitoSug
                                 mockStorageIface,
                                 fileLocs(2).atLine(1000).withTimestamp(lastModified),
                                 taskId,
-                                Option.empty,
       )
 
     val order = inOrder(batchListerFn)
@@ -173,54 +175,4 @@ class CloudSourceFileQueueTest extends AnyFlatSpec with Matchers with MockitoSug
     sourceFileQueue.next() shouldBe expected
   }
 
-  "list" should "not pass lastSeenFile to batchListerFn when PostProcessAction is set" in {
-
-    val batchListerFn =
-      mock[Option[TestFileMetadata] => Either[FileListError, Option[ListOfKeysResponse[TestFileMetadata]]]]
-
-    val mockStorageIface = mock[StorageInterface[TestFileMetadata]]
-    when(
-      mockStorageIface.seekToFile(
-        anyString(),
-        anyString(),
-        any[Option[Instant]],
-      ),
-    ).thenAnswer((_: String, file: String, _: Option[Instant]) => TestFileMetadata(file, lastModified).some)
-
-    val sourceFileQueue =
-      CloudSourceFileQueue.from(
-        batchListerFn,
-        mockStorageIface,
-        fileLocs(1).atLine(1000).withTimestamp(lastModified),
-        taskId,
-        Some(new DeletePostProcessAction(true)),
-      )
-
-    doAnswer(x => listBatch(x)).when(batchListerFn)(
-      any[Option[TestFileMetadata]],
-    )
-
-    val order = inOrder(batchListerFn)
-
-    // we are starting from a previously read file, so we use fileLocs(1) - this doesn't cause a read to the storage layer to discover next file
-    sourceFileQueue.next() should be(Right(Some(fileLocs(1).atLine(1000).withTimestamp(lastModified))))
-    order.verifyNoMoreInteractions()
-
-    verifyRefreshAndRollover(batchListerFn, sourceFileQueue, order)
-    verifyRefreshAndRollover(batchListerFn, sourceFileQueue, order)
-    verifyRefreshAndRollover(batchListerFn, sourceFileQueue, order)
-
-  }
-
-  private def verifyRefreshAndRollover(
-    batchListerFn:   Option[TestFileMetadata] => Either[FileListError, Option[ListOfKeysResponse[TestFileMetadata]]],
-    sourceFileQueue: CloudSourceFileQueue[TestFileMetadata],
-    order:           InOrder,
-  ): Unit = {
-    // REFRESH
-    sourceFileQueue.next() should be(Right(Some(fileLocs(0).atLine(-1).withTimestamp(lastModified))))
-    order.verify(batchListerFn)(none)
-    sourceFileQueue.next() should be(Right(Some(fileLocs(1).atLine(-1).withTimestamp(lastModified))))
-    order.verifyNoMoreInteractions()
-  }
 }