#637 remove usage of selectOffset

evolution-gaming · Aug 29, 2024 · 10be874 · 10be874
1 parent 23fe01b
commit 10be874
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 49 deletions.
diff --git a/...c/main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/PointerStatements.scala b/...c/main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/PointerStatements.scala
@@ -137,37 +137,4 @@ object PointerStatements {
         }
     }
   }
-
-  trait SelectOffset[F[_]] {
-
-    def apply(topic: Topic, partition: Partition): F[Option[Offset]]
-  }
-
-  object SelectOffset {
-
-    def of[F[_]: Monad: CassandraSession](
-      name: TableName,
-      consistencyConfig: EventualCassandraConfig.ConsistencyConfig.Read,
-    ): F[SelectOffset[F]] = {
-
-      val query =
-        s"""
-           |SELECT offset FROM ${name.toCql}
-           |WHERE topic = ?
-           |AND partition = ?
-           |""".stripMargin
-
-      query
-        .prepare
-        .map { prepared => (topic: Topic, partition: Partition) =>
-          prepared
-            .bind()
-            .encode("topic", topic)
-            .encode("partition", partition)
-            .setConsistencyLevel(consistencyConfig.value)
-            .first
-            .map { _.map { _.decode[Offset]("offset") } }
-        }
-    }
-  }
 }
diff --git a/...main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/ReplicatedCassandra.scala b/...main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/ReplicatedCassandra.scala
@@ -88,19 +88,7 @@ object ReplicatedCassandra {
                     new Main with ReplicatedPartitionJournal.Offsets[F] {
 
                       def get: F[Option[Offset]] = {
-                        for {
-                          offset <- statements.selectOffset2(topic, partition)
-                          offset <- offset.fold {
-                            val meterAndLog = {
-                              for {
-                                _      <- metrics.traverse_(_.selectOffsetFallback(topic, partition))
-                                message = s"`selectOffset` was called for topic $topic and partition: $partition"
-                                _      <- log.traverse_(_.warn(message))
-                              } yield ()
-                            }
-                            statements.selectOffset(topic, partition) <* meterAndLog
-                          } { _.some.pure[F] }
-                        } yield offset
+                        statements.selectOffset2(topic, partition)
                       }
 
                       def create(offset: Offset, timestamp: Instant): F[Unit] = {
@@ -705,7 +693,6 @@ object ReplicatedCassandra {
     deleteRecordsTo: JournalStatements.DeleteTo[F],
     deleteRecords: JournalStatements.Delete[F],
     metaJournal: MetaJournalStatements[F],
-    selectOffset: PointerStatements.SelectOffset[F],
     selectOffset2: Pointer2Statements.SelectOffset[F],
     selectPointer: PointerStatements.Select[F],
     selectPointer2: Pointer2Statements.Select[F],
@@ -732,7 +719,6 @@ object ReplicatedCassandra {
         deleteRecordsTo       <- JournalStatements.DeleteTo.of[F](schema.journal, consistencyConfig.write)
         deleteRecords         <- JournalStatements.Delete.of[F](schema.journal, consistencyConfig.write)
         metaJournal           <- MetaJournalStatements.of[F](schema, consistencyConfig)
-        selectOffset          <- PointerStatements.SelectOffset.of[F](schema.pointer, consistencyConfig.read)
         selectOffset2         <- Pointer2Statements.SelectOffset.of[F](schema.pointer2, consistencyConfig.read)
         selectPointer         <- PointerStatements.Select.of[F](schema.pointer, consistencyConfig.read)
         selectPointer2        <- Pointer2Statements.Select.of[F](schema.pointer2, consistencyConfig.read)
@@ -748,7 +734,6 @@ object ReplicatedCassandra {
           deleteRecordsTo,
           deleteRecords,
           metaJournal,
-          selectOffset,
           selectOffset2,
           selectPointer,
           selectPointer2,