diff --git a/journal/src/main/scala/com/evolution/kafka/journal/HeadCache.scala b/journal/src/main/scala/com/evolution/kafka/journal/HeadCache.scala index 33ee17db0..a495f67e8 100644 --- a/journal/src/main/scala/com/evolution/kafka/journal/HeadCache.scala +++ b/journal/src/main/scala/com/evolution/kafka/journal/HeadCache.scala @@ -161,8 +161,22 @@ object HeadCache { val log1 = log.prefixed(topic) cache .getOrUpdateResource(topic) { + val onBackgroundFailure = (e: Throwable) => + log1.error(s"background task failed, rebuilding cache: $e", e) >> + // Detach the eviction: `remove` releases this very resource (cancelling the + // task that called us), so it must not run within this scope. + cache.remove(topic).void.start.void TopicCache - .make(eventual, topic, log1, consumer, config, consRecordToActionHeader, metrics.map { _.headCache }) + .make( + eventual, + topic, + log1, + consumer, + config, + consRecordToActionHeader, + metrics.map { _.headCache }, + onBackgroundFailure, + ) .map { cache => metrics .fold(cache) { metrics => cache.withMetrics(topic, metrics.headCache) } diff --git a/journal/src/main/scala/com/evolution/kafka/journal/TopicCache.scala b/journal/src/main/scala/com/evolution/kafka/journal/TopicCache.scala index 7e6b194d4..7fe5c4c5f 100644 --- a/journal/src/main/scala/com/evolution/kafka/journal/TopicCache.scala +++ b/journal/src/main/scala/com/evolution/kafka/journal/TopicCache.scala @@ -70,6 +70,10 @@ private[journal] object TopicCache { * payload will be ignored. * @param metrics * Interface to report the metrics to. + * @param onBackgroundFailure + * Called once if any of the background tasks (Kafka consumption, Cassandra polling) terminates + * abnormally. The intended use is to invalidate this cache so it gets rebuilt on the next + * access, rather than silently degrading. It is not called on normal shutdown (cancellation). * @return * Resource which will configure a [[TopicCache]] with the passed parameters. Instance of * `Resource[TopicCache]` are, obviously, reusable and there is no need to call @@ -83,6 +87,7 @@ private[journal] object TopicCache { config: HeadCacheConfig, consRecordToActionHeader: ConsRecordToActionHeader[F], metrics: Option[HeadCache.Metrics[F]], + onBackgroundFailure: Throwable => F[Unit], ): Resource[F, TopicCache[F]] = { for { @@ -107,6 +112,19 @@ private[journal] object TopicCache { cachesMap = caches.toMap + // Completed with the first background task failure. A background task is not expected to + // terminate on its own: consumption retries forever and the polling loops are `foreverM`, so + // any completion (error or unexpected success) means the cache stopped doing its job. + failure <- Deferred[F, Throwable].toResource + onBackgroundExit = (name: String) => + (outcome: Outcome[F, Throwable, Unit]) => + outcome match { + case Outcome.Succeeded(_) => + failure.complete(JournalError(s"headcache background task '$name' terminated unexpectedly")).void + case Outcome.Errored(e) => failure.complete(e).void + case Outcome.Canceled() => ().pure[F] + } + remove = caches .foldMapM { case (partition, cache) => @@ -170,7 +188,12 @@ private[journal] object TopicCache { cachesMap .get(partition) .fold { - JournalError(s"invalid partition: $partition").raiseError[F, Sample] + // A partition appeared that was not present when the cache was built + // (the topic was repartitioned after init). `cachesMap` is fixed, so + // retrying cannot recover: signal failure to rebuild the whole cache, + // which re-reads partitions and picks up the new one. + val error = JournalError(s"invalid partition: $partition") + failure.complete(error) >> error.raiseError[F, Sample] } { cache => cache .add(records) @@ -213,7 +236,8 @@ private[journal] object TopicCache { } } yield result } - .onError { case e => log.error(s"consuming failed with $e", e) } /*TODO headcache: fail head cache*/ + .onError { case e => log.error(s"consuming failed with $e", e) } + .guaranteeCase(onBackgroundExit("consuming")) .background random <- Random.State.fromClock[F]().toResource strategy = Strategy @@ -226,7 +250,10 @@ private[journal] object TopicCache { .retry(strategy) .handleErrorWith { e => log.error(s"remove failed, error: $e", e) } .foreverM[Unit] + .guaranteeCase(onBackgroundExit("remove")) .background + // Rebuild the cache when a background task dies, instead of serving stale/empty results forever. + _ <- failure.get.flatMap { e => onBackgroundFailure(e) }.background _ <- metrics.foldMapM { metrics => val result = for { _ <- Temporal[F].sleep(1.minute) diff --git a/journal/src/test/scala/com/evolution/kafka/journal/HeadCacheSpec.scala b/journal/src/test/scala/com/evolution/kafka/journal/HeadCacheSpec.scala index 9623b1aaf..db7daba37 100644 --- a/journal/src/test/scala/com/evolution/kafka/journal/HeadCacheSpec.scala +++ b/journal/src/test/scala/com/evolution/kafka/journal/HeadCacheSpec.scala @@ -25,6 +25,13 @@ import scala.util.control.NoStackTrace class HeadCacheSpec extends AsyncWordSpec with Matchers { import HeadCacheSpec.* + private def eventually[A](fa: IO[A], attempts: Int = 250, delay: FiniteDuration = 20.millis): IO[A] = { + fa.handleErrorWith { e => + if (attempts <= 0) IO.raiseError(e) + else Temporal[IO].sleep(delay) >> eventually(fa, attempts - 1, delay) + } + } + "HeadCache" should { "return result, records are in cache" in { @@ -185,6 +192,52 @@ class HeadCacheSpec extends AsyncWordSpec with Matchers { result.run() } + "rebuild the cache when the topic is repartitioned" in { + val partition1 = Partition.unsafe(1) + val topicPartition1 = TopicPartition(topic = topic, partition = partition1) + val key0 = Key(id = "id0", topic = topic) + val key1 = Key(id = "id1", topic = topic) + + def recordOn(tp: TopicPartition, key: Key): ConsumerRecords[String, Unit] = { + ConsumerRecordsOf(List(consumerRecordOf(appendOf(key, SeqNr.min), tp, Offset.min))) + } + + def assignedPartitions(state: TestConsumer.State): Set[Partition] = { + state + .actions + .collect { case TestConsumer.Action.Assign(_, partitions) => partitions.toSortedSet } + .foldLeft(Set.empty[Partition]) { _ ++ _ } + } + + val state = TestConsumer.State(topics = Map((topic, List(partition)))) + + val result = for { + stateRef <- Ref[IO].of(state) + consumer = TestConsumer.make(stateRef) + _ <- headCacheOf(HeadCache.Eventual.empty, consumer).use { headCache => + for { + // build the per-topic cache while only partition 0 exists + _ <- stateRef.update { _.enqueue(recordOn(topicPartition, key0).pure[Try]) } + a <- headCache.get(key0, partition, Offset.min) + _ = a shouldEqual HeadInfo.append(Offset.min, SeqNr.min, none).some + // topic gets repartitioned: partition 1 now exists, and a record lands on it + _ <- stateRef.update { _.copy(topics = Map((topic, List(partition, partition1)))) } + _ <- stateRef.update { _.enqueue(recordOn(topicPartition1, key1).pure[Try]) } + // The unknown partition makes the cache fail and evict itself; the next `get` rebuilds + // it, re-reading partitions and this time assigning the new partition 1 (instead of + // looping forever). `get` drives the lazy rebuild; errors/None during recovery are + // tolerated until the assign is observed. + _ <- eventually { + headCache.get(key1, partition1, Offset.min).attempt >> + stateRef.get.map { s => assignedPartitions(s) should contain(partition1) } + } + } yield {} + } + } yield {} + + result.run(10.seconds) + } + "timeout" in { val headCache = headCacheOf(HeadCache.Eventual.empty, consumerEmpty.pure[IO].toResource, config.copy(timeout = 10.millis))