Prepare CassandraJournals statements only once at the startup (#739)

Author: Z1kkurat

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/CassandraSessionStub.scala

@@ -7,7 +7,7 @@ import com.datastax.driver.core.{Host, PreparedStatement, RegularStatement, Resu
 import com.evolutiongaming.scassandra.CassandraSession
 
 object CassandraSessionStub {
-
+  // Doesn't inject failures on statements preparation since we don't prepare them on each call.
   def injectFailures[F[_]](
     session: CassandraSession[F],
     failAfter: Ref[F, Int]
@@ -24,49 +24,6 @@ object CassandraSessionStub {
     override def init: F[Unit]                        = F.unit
     override def execute(query: String): F[ResultSet] = failed.ifM(fail(query), session.execute(query))
 
-    override def execute(query: String, values: Any*): F[ResultSet] =
-      failed.ifM(fail(query), session.execute(query, values: _*))
-
-    override def execute(query: String, values: Map[String, AnyRef]): F[ResultSet] =
-      failed.ifM(fail(query), session.execute(query, values))
-
-    override def execute(statement: Statement): F[ResultSet] =
-      failed.ifM(fail(statement.toString), session.execute(statement))
-
-    override def prepare(query: String): F[PreparedStatement] =
-      failed.ifM(fail(query), session.prepare(query))
-
-    override def prepare(statement: RegularStatement): F[PreparedStatement] =
-      failed.ifM(fail(statement.toString), session.prepare(statement))
-
-    override def state: CassandraSession.State[F] = new CassandraSession.State[F] {
-      override def connectedHosts: F[Iterable[Host]]      = F.pure(Iterable.empty)
-      override def openConnections(host: Host): F[Int]    = F.pure(0)
-      override def trashedConnections(host: Host): F[Int] = F.pure(0)
-      override def inFlightQueries(host: Host): F[Int]    = F.pure(0)
-    }
-
-  }
-
-  // Doesn't inject failures on statements preparation since we want to avoid preparing them on each call.
-  // This is a temporary method to be used in specs that test the classes that have already been migrated to
-  // preparing statements in advance
-  def injectFailuresOnExecute[F[_]](
-    session: CassandraSession[F],
-    failAfter: Ref[F, Int]
-  )(implicit F: MonadThrow[F]): CassandraSession[F] = new CassandraSession[F] {
-    def fail[T](query: String): F[T] = F.raiseError {
-      new RuntimeException(s"CassandraSessionStub: failing after proper calls exhausted: $query")
-    }
-
-    val failed = failAfter modify { failAfter =>
-      (failAfter - 1, failAfter <= 0)
-    }
-
-    override def loggedKeyspace: F[Option[String]]    = F.pure(None)
-    override def init: F[Unit]                        = F.unit
-    override def execute(query: String): F[ResultSet] = failed.ifM(fail(query), session.execute(query))
-
     override def execute(query: String, values: Any*): F[ResultSet] =
       failed.ifM(fail(query), session.execute(query, values: _*))
 

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala

@@ -20,7 +20,7 @@ import scala.concurrent.duration.*
 class FlowSpec extends CassandraSpec {
 
   test("flow fails when Cassandra insert fails") {
-    val flow = for {
+    val flow: Resource[IO, IO[Unit]] = for {
       failAfter <- Resource.eval(Ref.of[IO, Int](10000))
       session    = CassandraSessionStub.injectFailures(cassandra().session, failAfter)
       storage <- Resource.eval(
@@ -78,11 +78,7 @@ class FlowSpec extends CassandraSpec {
       }
     } yield join
 
-    val test: IO[Unit] = flow use { join =>
-      join.attempt map { result =>
-        assert(clue(result.isLeft))
-      }
-    }
+    val test: IO[Unit] = flow.use(join => join.attempt.map(result => assert(clue(result.isLeft))))
 
     test.unsafeRunSync()
   }

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/journal/JournalSpec.scala

@@ -47,7 +47,7 @@ class JournalSpec extends CassandraSpec {
       failAfter <- Ref.of[IO, Int](100)
       session    = CassandraSessionStub.injectFailures(cassandra().session, failAfter)
       journals  <- CassandraJournals.withSchema(session, cassandra().sync)
-      _         <- failAfter.set(1)
+      _         <- failAfter.set(0) // fail immediately on the first read attempt
       records   <- journals.get(key).toList.attempt
     } yield assert(clue(records.isLeft))
 

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/key/KeySpec.scala

@@ -52,7 +52,7 @@ class KeySpec extends CassandraSpec {
   test("failures") {
     val test: IO[Unit] = for {
       failAfter <- Ref.of[IO, Int](100)
-      session    = CassandraSessionStub.injectFailuresOnExecute(cassandra().session, failAfter)
+      session    = CassandraSessionStub.injectFailures(cassandra().session, failAfter)
       keys <- CassandraKeys.withSchema(
         session,
         cassandra().sync,

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotSpec.scala

@@ -36,7 +36,7 @@ class SnapshotSpec extends CassandraSpec {
     val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "queries")
     val test: IO[Unit] = for {
       failAfter <- Ref.of[IO, Int](100)
-      session    = CassandraSessionStub.injectFailuresOnExecute(cassandra().session, failAfter)
+      session    = CassandraSessionStub.injectFailures(cassandra().session, failAfter)
       snapshots <- CassandraSnapshots.withSchema[IO, String](session, cassandra().sync)
       _         <- failAfter.set(0) // fail immediately on the first read attempt
       snapshots <- snapshots.get(key).attempt