Removed all debugging code because the bug is fixed now (#131).

evolution-gaming · Jan 10, 2021 · cddb626 · cddb626
1 parent 3e8a105
commit cddb626
Show file tree

Hide file tree

Showing 8 changed files with 12 additions and 40 deletions.
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/ConsumerFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/ConsumerFlow.scala
@@ -31,7 +31,7 @@ object ConsumerFlow {
     * Note, that topic specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def of[F[_]: MonadThrowable: Timeout: LogOf](
+  def of[F[_]: MonadThrowable: LogOf](
     consumer: Consumer[F],
     topic: Topic,
     flowOf: TopicFlowOf[F],
@@ -48,7 +48,7 @@ object ConsumerFlow {
     * Note, that topics specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def of[F[_]: MonadThrowable: Timeout: LogOf](
+  def of[F[_]: MonadThrowable: LogOf](
     consumer: Consumer[F],
     topics: NonEmptySet[Topic],
     flowOf: TopicFlowOf[F],
@@ -68,7 +68,7 @@ object ConsumerFlow {
     * Note, that topic specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def apply[F[_]: MonadThrowable: Timeout: LogOf](
+  def apply[F[_]: MonadThrowable: LogOf](
     consumer: Consumer[F],
     flows: Map[Topic, TopicFlow[F]],
     config: ConsumerFlowConfig
@@ -92,7 +92,7 @@ object ConsumerFlow {
 
     def stream = for {
       _ <- Stream.lift(subscribe)
-      records <- Stream.repeat(Timeout[F].timeout(poll))
+      records <- Stream.repeat(poll)
       // we process empty polls to trigger timers, but do not return them
       if records.values.nonEmpty
     } yield records

diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/ConsumerFlowOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/ConsumerFlowOf.scala
@@ -20,7 +20,7 @@ object ConsumerFlowOf {
     * Note, that topic specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def apply[F[_]: BracketThrowable: Timeout: LogOf](
+  def apply[F[_]: BracketThrowable: LogOf](
     topic: Topic,
     flowOf: TopicFlowOf[F],
     config: ConsumerFlowConfig = ConsumerFlowConfig()
@@ -33,7 +33,7 @@ object ConsumerFlowOf {
     * Note, that topics specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def apply[F[_]: BracketThrowable: Timeout: LogOf](
+  def apply[F[_]: BracketThrowable: LogOf](
     topics: NonEmptySet[Topic],
     flowOf: TopicFlowOf[F],
   ): ConsumerFlowOf[F] = ConsumerFlowOf(topics, flowOf, ConsumerFlowConfig())
@@ -43,7 +43,7 @@ object ConsumerFlowOf {
     * Note, that topics specified by an appropriate parameter should contain a
     * journal in the format of `Kafka Journal` library.
     */
-  def apply[F[_]: BracketThrowable: Timeout: LogOf](
+  def apply[F[_]: BracketThrowable: LogOf](
     topics: NonEmptySet[Topic],
     flowOf: TopicFlowOf[F],
     config: ConsumerFlowConfig

diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/Timeout.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/Timeout.scala
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/ConsumerFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/ConsumerFlowSpec.scala
@@ -87,7 +87,6 @@ object ConsumerFlowSpec {
   type F[A] = StateT[SyncIO, Context, A]
 
   implicit val logOf: LogOf[F] = LogOf.empty
-  implicit val timeout: Timeout[F] = Timeout.never
 
   case class StopException(context: Context) extends Exception
 

diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/KafkaFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/KafkaFlowSpec.scala
@@ -278,6 +278,4 @@ object KafkaFlowSpec {
 
   case object Error extends RuntimeException with NoStackTrace
 
-  implicit val timeout: Timeout[F] = Timeout.never
-
 }
diff --git a/docs/overview.md b/docs/overview.md
@@ -70,7 +70,6 @@ implicit val bracketThrowable = BracketThrowable[IO]
 implicit val measureDuration = MeasureDuration.empty[IO]
 implicit val logOf = LogOf.empty[IO]
 implicit val log = Log.empty[IO]
-implicit val timetout = Timeout.never[IO]
 ```
 
 ## KafkaFlow

diff --git a/persistence-cassandra/src/it/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala b/persistence-cassandra/src/it/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala
@@ -64,7 +64,6 @@ class FlowSpec(val globalResources: GlobalResources) extends CassandraSpec {
        ConsumerRecords(Map(TopicPartition.empty -> records))
      }
      join <- {
-        implicit val timeout = Timeout.of(10.minutes)
         implicit val retry = Retry.empty[IO]
         KafkaFlow.resource(
           consumer = Resource.liftF(consumer),

diff --git a/...e-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraModule.scala b/...e-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraModule.scala
@@ -27,8 +27,6 @@ object CassandraModule {
 
   def log[F[_]: LogOf]: F[Log[F]] = LogOf[F].apply(CassandraModule.getClass)
 
-  val FutureTimeout: FiniteDuration = 5.minutes
-
   def clusterOf[F[_]: Sync](
     fromGFuture: FromGFuture[F]
   ): F[CassandraClusterOf[F]] = {
@@ -49,13 +47,16 @@ object CassandraModule {
 
     for {
       log             <- Resource.liftF(log[F])
+      // this is required to log all Cassandra errors before popping them up,
+      // which is useful because popped up errors might be lost in some cases
+      // while kafka-flow is accessing Cassandra in bracket/resource release
+      // routine
       fromGFuture     = new FromGFuture[F] {
         val self = FromGFuture.lift[F]
         def apply[A](future: => ListenableFuture[A]) = {
-          val fa = self(future) onError { case e =>
+          self(future) onError { case e =>
             log.error("Cassandra request failed", e)
           }
-          Concurrent.timeout(fa, FutureTimeout)
         }
       }
       clusterOf       <- Resource.liftF(clusterOf[F](fromGFuture))