<commit_before>package mesosphere.marathon
package storage.repository

import java.time.OffsetDateTime
import java.time.format.DateTimeFormatter

import akka.http.scaladsl.marshalling.Marshaller
import akka.http.scaladsl.unmarshalling.Unmarshaller
import akka.stream.Materializer
import akka.stream.scaladsl.Source
import akka.{ Done, NotUsed }
import com.typesafe.scalalogging.StrictLogging
import mesosphere.marathon.core.pod.PodDefinition
import mesosphere.marathon.core.storage.repository.impl.PersistenceStoreVersionedRepository
import mesosphere.marathon.core.storage.store.impl.BasePersistenceStore
import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LazyVersionCachingPersistentStore, LoadTimeCachingPersistenceStore }
import mesosphere.marathon.core.storage.store.{ IdResolver, PersistenceStore }
import mesosphere.marathon.state.{ AppDefinition, Group, RootGroup, PathId, Timestamp }
import mesosphere.marathon.stream.Implicits._
import mesosphere.marathon.util.{ RichLock, toRichFuture }

import scala.annotation.tailrec
import scala.async.Async.{ async, await }
import scala.collection.concurrent.TrieMap
import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.control.NonFatal
import scala.util.{ Failure, Success }

case class StoredGroup(
    id: PathId,
    appIds: Map[PathId, OffsetDateTime],
    podIds: Map[PathId, OffsetDateTime],
    storedGroups: Seq[StoredGroup],
    dependencies: Set[PathId],
    version: OffsetDateTime) extends StrictLogging {

  lazy val transitiveAppIds: Map[PathId, OffsetDateTime] = appIds ++ storedGroups.flatMap(_.appIds)
  lazy val transitivePodIds: Map[PathId, OffsetDateTime] = podIds ++ storedGroups.flatMap(_.podIds)

  @SuppressWarnings(Array("all")) // async/await
  def resolve(
    appRepository: AppRepository,
    podRepository: PodRepository)(implicit ctx: ExecutionContext): Future[Group] = async { // linter:ignore UnnecessaryElseBranch
    val appFutures = appIds.map {
      case (appId, appVersion) => appRepository.getVersion(appId, appVersion).recover {
        case NonFatal(ex) =>
          logger.error(s"Failed to load $appId:$appVersion for group $id ($version)", ex)
          None
      }
    }
    val podFutures = podIds.map {
      case (podId, podVersion) => podRepository.getVersion(podId, podVersion).recover {
        case NonFatal(ex) =>
          logger.error(s"Failed to load $podId:$podVersion for group $id ($version)", ex)
          None
      }
    }

    val groupFutures = storedGroups.map(_.resolve(appRepository, podRepository))

    val allApps = await(Future.sequence(appFutures))
    if (allApps.exists(_.isEmpty)) {
      logger.warn(s"Group $id $version is missing ${allApps.count(_.isEmpty)} apps")
    }

    val allPods = await(Future.sequence(podFutures))
    if (allPods.exists(_.isEmpty)) {
      logger.warn(s"Group $id $version is missing ${allPods.count(_.isEmpty)} pods")
    }

    val apps: Map[PathId, AppDefinition] = await(Future.sequence(appFutures)).collect {
      case Some(app: AppDefinition) =>
        app.id -> app
    }(collection.breakOut)

    val pods: Map[PathId, PodDefinition] = await(Future.sequence(podFutures)).collect {
      case Some(pod: PodDefinition) =>
        pod.id -> pod
    }(collection.breakOut)

    val groups: Map[PathId, Group] = await(Future.sequence(groupFutures)).map(group => group.id -> group)(collection.breakOut)

    Group(
      id = id,
      apps = apps,
      pods = pods,
      groupsById = groups,
      dependencies = dependencies,
      version = Timestamp(version),
      transitiveAppsById = apps ++ groups.values.flatMap(_.transitiveAppsById),
      transitivePodsById = pods ++ groups.values.flatMap(_.transitivePodsById)
    )
  }

  def toProto: Protos.GroupDefinition = {
    import StoredGroup.DateFormat

    val b = Protos.GroupDefinition.newBuilder
      .setId(id.safePath)
      .setVersion(DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(version))

    appIds.foreach {
      case (app, appVersion) =>
        b.addApps(
          Protos.GroupDefinition.AppReference.newBuilder()
            .setId(app.safePath)
            .setVersion(DateFormat.format(appVersion)))
    }<commit_msg>Do not ignore exceptions when resolving apps and pods in GroupRepository (#5599)<commit_after>package mesosphere.marathon
package storage.repository

import java.time.OffsetDateTime
import java.time.format.DateTimeFormatter

import akka.http.scaladsl.marshalling.Marshaller
import akka.http.scaladsl.unmarshalling.Unmarshaller
import akka.stream.Materializer
import akka.stream.scaladsl.Source
import akka.{ Done, NotUsed }
import com.typesafe.scalalogging.StrictLogging
import mesosphere.marathon.core.pod.PodDefinition
import mesosphere.marathon.core.storage.repository.impl.PersistenceStoreVersionedRepository
import mesosphere.marathon.core.storage.store.impl.BasePersistenceStore
import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LazyVersionCachingPersistentStore, LoadTimeCachingPersistenceStore }
import mesosphere.marathon.core.storage.store.{ IdResolver, PersistenceStore }
import mesosphere.marathon.state.{ AppDefinition, Group, RootGroup, PathId, Timestamp }
import mesosphere.marathon.stream.Implicits._
import mesosphere.marathon.util.{ RichLock, toRichFuture }

import scala.annotation.tailrec
import scala.async.Async.{ async, await }
import scala.collection.concurrent.TrieMap
import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.control.NonFatal
import scala.util.{ Failure, Success }

case class StoredGroup(
    id: PathId,
    appIds: Map[PathId, OffsetDateTime],
    podIds: Map[PathId, OffsetDateTime],
    storedGroups: Seq[StoredGroup],
    dependencies: Set[PathId],
    version: OffsetDateTime) extends StrictLogging {

  lazy val transitiveAppIds: Map[PathId, OffsetDateTime] = appIds ++ storedGroups.flatMap(_.appIds)
  lazy val transitivePodIds: Map[PathId, OffsetDateTime] = podIds ++ storedGroups.flatMap(_.podIds)

  @SuppressWarnings(Array("all")) // async/await
  def resolve(
    appRepository: AppRepository,
    podRepository: PodRepository)(implicit ctx: ExecutionContext): Future[Group] = async { // linter:ignore UnnecessaryElseBranch
    val appFutures = appIds.map {
      case (appId, appVersion) => appRepository.getVersion(appId, appVersion).recover {
        case NonFatal(ex) =>
          logger.error(s"Failed to load $appId:$appVersion for group $id ($version)", ex)
          throw ex
      }
    }
    val podFutures = podIds.map {
      case (podId, podVersion) => podRepository.getVersion(podId, podVersion).recover {
        case NonFatal(ex) =>
          logger.error(s"Failed to load $podId:$podVersion for group $id ($version)", ex)
          throw ex
      }
    }

    val groupFutures = storedGroups.map(_.resolve(appRepository, podRepository))

    val allApps = await(Future.sequence(appFutures))
    if (allApps.exists(_.isEmpty)) {
      logger.warn(s"Group $id $version is missing ${allApps.count(_.isEmpty)} apps")
    }

    val allPods = await(Future.sequence(podFutures))
    if (allPods.exists(_.isEmpty)) {
      logger.warn(s"Group $id $version is missing ${allPods.count(_.isEmpty)} pods")
    }

    val apps: Map[PathId, AppDefinition] = await(Future.sequence(appFutures)).collect {
      case Some(app: AppDefinition) =>
        app.id -> app
    }(collection.breakOut)

    val pods: Map[PathId, PodDefinition] = await(Future.sequence(podFutures)).collect {
      case Some(pod: PodDefinition) =>
        pod.id -> pod
    }(collection.breakOut)

    val groups: Map[PathId, Group] = await(Future.sequence(groupFutures)).map { group =>
      group.id -> group
    }(collection.breakOut)

    Group(
      id = id,
      apps = apps,
      pods = pods,
      groupsById = groups,
      dependencies = dependencies,
      version = Timestamp(version),
      transitiveAppsById = apps ++ groups.values.flatMap(_.transitiveAppsById),
      transitivePodsById = pods ++ groups.values.flatMap(_.transitivePodsById)
    )
  }

  def toProto: Protos.GroupDefinition = {
    import StoredGroup.DateFormat

    val b = Protos.GroupDefinition.newBuilder
      .setId(id.safePath)
      .setVersion(DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(version))

    appIds.foreach {
      case (app, appVersion) =>
        b.addApps(
          Protos.GroupDefinition.AppReference.newBuilder()
            .setId(app.safePath)
            .setVersion(DateFormat.format(appVersion)))
    }
{'fraction_non_alphanumeric': 0.10789597000937207, 'fraction_numerical': 0.00046860356138706655, 'mean_word_length': 4.007038123167155, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5530800', 'n_tokens_mistral': 2634, 'n_tokens_neox': 2563, 'n_words': 515}