diff --git a/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala similarity index 84% rename from hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala rename to common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala index 8343bb887e..2177bb3821 100644 --- a/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala +++ b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala @@ -8,18 +8,31 @@ import akka.cluster.sharding.typed.scaladsl.{ClusterSharding, EntityRef} import akka.persistence.typed.PersistenceId import akka.persistence.typed.scaladsl.EventSourcedBehavior.CommandHandler import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior, RetentionCriteria} +import akka.actor.typed.ActorSystem +import akka.actor.typed.scaladsl.adapter._ +import akka.actor.{ActorSystem => ClassicActorSystem} +import akka.management.cluster.bootstrap.ClusterBootstrap +import akka.management.scaladsl.AkkaManagement import akka.stream.Materializer -import com.typesafe.config.Config -import hmda.HmdaPlatform.{institutionKafkaProducer, log} +import akka.util.Timeout + +import scala.concurrent.ExecutionContext +import org.slf4j.LoggerFactory + +import scala.concurrent.duration._ +import akka.util.Timeout +import com.typesafe.config.{Config, ConfigFactory} import hmda.messages.institution.InstitutionCommands._ import hmda.messages.institution.InstitutionEvents._ import hmda.messages.pubsub.HmdaTopics._ import hmda.model.institution.{Institution, InstitutionDetail} import hmda.persistence.HmdaTypedPersistentActor +import hmda.publication.KafkaUtils import hmda.publication.KafkaUtils._ -import com.typesafe.config.ConfigFactory +import hmda.util.FieldNullifyUtility.nullifyInstitutionFields +import org.apache.kafka.clients.producer.{KafkaProducer, Producer} -import scala.concurrent.Future +import scala.concurrent.{ExecutionContext, Future} object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionCommand, InstitutionEvent, InstitutionState] { override final val name = "Institution" @@ -27,6 +40,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman override def behavior(entityId: String): Behavior[InstitutionCommand] = Behaviors.setup { ctx => ctx.log.info(s"Started Institution: $entityId") + val sharding = ClusterSharding(ctx.system) EventSourcedBehavior[InstitutionCommand, InstitutionEvent, InstitutionState]( persistenceId = PersistenceId.ofUniqueId(entityId), emptyState = InstitutionState(None), @@ -46,6 +60,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman val louConfig = ConfigFactory.load("application.conf").getConfig("filter") val louList = louConfig.getString("lou-filter-list").toUpperCase.split(",") + val institutionKafkaProducer = KafkaUtils.getInstitutionKafkaProducer(system) (state, cmd) => cmd match { @@ -60,7 +75,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman Effect.persist(InstitutionCreated(i)).thenRun { _ => log.debug(s"Institution Created: ${i.toString}") val event = InstitutionCreated(i) - publishInstitutionEvent(i.LEI, InstitutionKafkaEvent("InstitutionCreated", event), config) + publishInstitutionEvent(i.LEI, InstitutionKafkaEvent("InstitutionCreated", event), config,institutionKafkaProducer) replyTo ! event } } else { @@ -77,7 +92,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman Effect.persist(InstitutionModified(i)).thenRun { _ => log.debug(s"Institution Modified: ${i.toString}") val event = InstitutionModified(i) - publishInstitutionEvent(i.LEI, InstitutionKafkaEvent("InstitutionModified", event), config) + publishInstitutionEvent(i.LEI, InstitutionKafkaEvent("InstitutionModified", event), config,institutionKafkaProducer) replyTo ! event } } else { @@ -93,7 +108,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman Effect.persist(InstitutionDeleted(lei, activityYear)).thenRun { _ => log.debug(s"Institution Deleted: $lei") val event = InstitutionDeleted(lei, activityYear) - publishInstitutionEvent(lei, InstitutionKafkaEvent("InstitutionDeleted", event), config) + publishInstitutionEvent(lei, InstitutionKafkaEvent("InstitutionDeleted", event), config,institutionKafkaProducer) replyTo ! event } } else { @@ -143,7 +158,8 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman private def publishInstitutionEvent( institutionID: String, event: InstitutionKafkaEvent, - config: Config + config: Config, + institutionKafkaProducer: Producer[String,InstitutionKafkaEvent] )(implicit system: ActorSystem[_], materializer: Materializer): Future[Done] = produceInstitutionRecord(institutionTopic, institutionID, event, institutionKafkaProducer) @@ -152,7 +168,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman state } else { if (institution.LEI == state.institution.get.LEI) { - state.copy(Some(institution)) + state.copy(Some(nullifyInstitutionFields(institution))) } else { state } @@ -169,7 +185,7 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman if(parts.size == 1) Some(EntityId(parts.head, 2018)) else if(parts.size == 2) Some(EntityId(parts(0), parts(1).toInt)) else { - log.error(s"Cant parse institution entity id: ${entityIdStr}") + // log.error(s"Cant parse institution entity id: ${entityIdStr}") None } } else { @@ -183,4 +199,5 @@ object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionComman def selectInstitution(sharding: ClusterSharding, lei: String, year: Int): EntityRef[InstitutionCommand] = sharding.entityRefFor(InstitutionPersistence.typeKey, makeEntityId(lei, year)) + } \ No newline at end of file diff --git a/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsProtobufConverter.scala b/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsProtobufConverter.scala index a9e7c2d57b..638deef1f3 100644 --- a/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsProtobufConverter.scala +++ b/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsProtobufConverter.scala @@ -9,6 +9,7 @@ import hmda.persistence.serialization.institution.InstitutionMessage import hmda.persistence.serialization.institution.commands._ import hmda.serialization.filing.FilingProtobufConverter._ import hmda.serialization.institution.InstitutionProtobufConverter._ +import hmda.util.FieldNullifyUtility.nullifyInstitutionFields // $COVERAGE-OFF$ object InstitutionCommandsProtobufConverter { @@ -22,7 +23,7 @@ object InstitutionCommandsProtobufConverter { val msg = CreateInstitutionMessage.parseFrom(bytes) val institution = institutionFromProtobuf(msg.institution.getOrElse(InstitutionMessage())) val actorRef = resolver.resolveActorRef(msg.replyTo) - CreateInstitution(institution, actorRef) + CreateInstitution(nullifyInstitutionFields(institution), actorRef) } def modifyInstitutionToProtobuf( @@ -41,7 +42,7 @@ object InstitutionCommandsProtobufConverter { val msg = ModifyInstitutionMessage.parseFrom(bytes) val institution = institutionFromProtobuf(msg.institution.getOrElse(InstitutionMessage())) val actorRef = resolver.resolveActorRef[InstitutionEvent](msg.replyTo) - ModifyInstitution(institution, actorRef) + ModifyInstitution(nullifyInstitutionFields(institution), actorRef) } def getInstitutionToProtobuf( diff --git a/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsSerializer.scala b/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsSerializer.scala index 7ac047626f..be2e1e00b0 100644 --- a/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsSerializer.scala +++ b/common/src/main/scala/hmda/serialization/institution/InstitutionCommandsSerializer.scala @@ -1,16 +1,16 @@ package hmda.serialization.institution import java.io.NotSerializableException - import akka.actor.ExtendedActorSystem import akka.actor.typed.ActorRefResolver import akka.actor.typed.scaladsl.adapter._ import akka.serialization.SerializerWithStringManifest -import hmda.messages.institution.InstitutionCommands.{ GetInstitutionDetails, _ } +import hmda.messages.institution.InstitutionCommands.{GetInstitutionDetails, _} import hmda.model.institution.Institution import hmda.persistence.serialization.institution.InstitutionMessage import hmda.serialization.institution.InstitutionCommandsProtobufConverter._ import hmda.serialization.institution.InstitutionProtobufConverter._ +import hmda.util.FieldNullifyUtility.nullifyInstitutionFields // $COVERAGE-OFF$ class InstitutionCommandsSerializer(system: ExtendedActorSystem) extends SerializerWithStringManifest { @@ -31,7 +31,7 @@ class InstitutionCommandsSerializer(system: ExtendedActorSystem) extends Seriali override def toBinary(o: AnyRef): Array[Byte] = o match { case i: Institution => - institutionToProtobuf(i).toByteArray + institutionToProtobuf(nullifyInstitutionFields(i)).toByteArray case cmd: CreateInstitution => createInstitutionToProtobuf(cmd, resolver).toByteArray case cmd: ModifyInstitution => diff --git a/data/2025/yearly/bank0/Bank0_clean_100_rows_2025_annual.txt b/data/2025/yearly/bank0/Bank0_clean_100_rows_2025_annual.txt index dfbc247474..a3bdc2d4d0 100644 --- a/data/2025/yearly/bank0/Bank0_clean_100_rows_2025_annual.txt +++ b/data/2025/yearly/bank0/Bank0_clean_100_rows_2025_annual.txt @@ -1,4 +1,4 @@ -1|Bank0|2025|4|Mr. Smug Pockets|555-555-5555|pockets@ficus.com|1234 Hocus Potato Way|Tatertown|UT|84096|9|100|01-0123456|B90YWS6AFX2LGWOXJ1LD +1|Ts Panel Updater 07/09 - 1 |2025|4|Mr. Smug Pockets|555-555-5555|pockets@ficus.com|1234 Hocus Potato Way|Tatertown|UT|84096|9|100|01-0123456|B90YWS6AFX2LGWOXJ1LD 2|B90YWS6AFX2LGWOXJ1LD|B90YWS6AFX2LGWOXJ1LDJ4JJTLR928LPCCRN5MMXN0904|20250113|3|2|2|2|3|218910|5|20251010|1234 Hocus Potato Way|Washington|DC|14755|11001|11001980000|1|13||11||KE0NW|1||||||2|2|7||||||||27|24|41|43|2||||3|2|2|2|1|1|75|44|85|0|NA|3|2|8888|8888|9||9||10|||||NA|NA|NA|NA|NA|NA|NA|NA|NA|256|29|2|2|2|2|NA|1|2|4|NA|2|2|NA|3|1|5|1||DOREBESQSW1QT58SD2OZTHQUGXLSKCAJYZ63NJE2MUIAFQL4KW6PU26YSU786GT0IMCWWKCN25Y7KU0VLU0PPKWR8G6DKWI9BANPIE9I2ZZ5XDUX0TBAY4XFRFQZF087WS9ESTAKIV5V9HSZ2VXW7J5JMGPP4CGYA51BK68T57NN4KTKJVXIQMFXBTN5E3LGKKX3LITQ4C7OPFJ|8|6|5|5|||2|2|1 2|B90YWS6AFX2LGWOXJ1LD|B90YWS6AFX2LGWOXJ1LD5GBVI7IK47Z2GOXLNZBLMZ689|20250113|3|2|2|2|3|218910|5|20251010|1234 Hocus Potato Way|Washington|DC|14755|11001|11001980000|1|13||11||KE0NW|1||||||2|2|7||||||||27|24|41|43|2||||3|2|2|2|1|1|75|44|85|0|NA|3|2|8888|8888|9||9||10|||||NA|NA|NA|NA|NA|NA|NA|NA|NA|256|29|2|2|2|2|NA|1|2|4|NA|2|2|NA|3|1|5|1||DOREBESQSW1QT58SD2OZTHQUGXLSKCAJYZ63NJE2MUIAFQL4KW6PU26YSU786GT0IMCWWKCN25Y7KU0VLU0PPKWR8G6DKWI9BANPIE9I2ZZ5XDUX0TBAY4XFRFQZF087WS9ESTAKIV5V9HSZ2VXW7J5JMGPP4CGYA51BK68T57NN4KTKJVXIQMFXBTN5E3LGKKX3LITQ4C7OPFJ|8|6|5|5|||2|2|1 2|B90YWS6AFX2LGWOXJ1LD|B90YWS6AFX2LGWOXJ1LDYUDA8ZEVVS3CXU58MODGOYE22|20250113|3|2|2|2|3|218910|5|20251010|1234 Hocus Potato Way|Washington|DC|14755|11001|11001980000|1|13||11||KE0NW|1||||||2|2|7||||||||27|24|41|43|2||||3|2|2|2|1|1|75|44|85|0|NA|3|2|8888|8888|9||9||10|||||NA|NA|NA|NA|NA|NA|NA|NA|NA|256|29|2|2|2|2|NA|1|2|4|NA|2|2|NA|3|1|5|1||DOREBESQSW1QT58SD2OZTHQUGXLSKCAJYZ63NJE2MUIAFQL4KW6PU26YSU786GT0IMCWWKCN25Y7KU0VLU0PPKWR8G6DKWI9BANPIE9I2ZZ5XDUX0TBAY4XFRFQZF087WS9ESTAKIV5V9HSZ2VXW7J5JMGPP4CGYA51BK68T57NN4KTKJVXIQMFXBTN5E3LGKKX3LITQ4C7OPFJ|8|6|5|5|||2|2|1 diff --git a/hmda-analytics/src/main/resources/application.conf b/hmda-analytics/src/main/resources/application.conf index 8f6e5ee3b3..59a570d2ac 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -5,10 +5,46 @@ # AWS Keyspace with SigV4AuthProvider # include "persistence-keyspace-sigv4.conf" +hmda.cluster.name = hmdaAnalytics + akka { - loglevel = info - log-dead-letters = off - log-dead-letters-during-shutdown = off + diagnostics { + starvation-detector { + check-interval = 1s + initial-delay = 30s + max-delay-warning-threshold = 100 ms + warning-interval = 10 seconds + } + } + + blocking-quality-dispatcher { + type = Dispatcher + executor = "fork-join-executor" + fork-join-executor { + parallelism-min = 2 + parallelism-factor = 2.0 + } + throughput = 1000 + } + actor { + provider = "cluster" + timeout = 8 + } + remote.artery { + canonical { + hostname = "127.0.0.1" + port = 2551 + } + } + + cluster { + seed-nodes = [ + "akka://"${hmda.cluster.name}"@127.0.0.1:2551" + ] + + downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider" + min-nr-of-members = 1 + } } cassandra-journal { @@ -44,6 +80,11 @@ cassandra-snapshot-store { } hmda { + supervisor { + minBackOff = 10 + maxBackOff = 60 + randomFactor = 0.1 + } analytics { parallelism = 1 tsDeletion = true @@ -57,6 +98,8 @@ hmda { 2026 { tsTableName = "transmittalsheet2026" tsTableName = ${?TS_TABLE_NAME_2026} + institutionTableName = "institutions2026" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2026} tsTableNameQ1 = "ts2026_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2026_Q1} tsTableNameQ2 = "ts2026_q2" @@ -75,6 +118,8 @@ hmda { 2025 { tsTableName = "transmittalsheet2025" tsTableName = ${?TS_TABLE_NAME_2025} + institutionTableName = "institutions2025" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2025} tsTableNameQ1 = "ts2025_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2025_Q1} tsTableNameQ2 = "ts2025_q2" @@ -93,6 +138,8 @@ hmda { 2024 { tsTableName = "transmittalsheet2024" tsTableName = ${?TS_TABLE_NAME_2024} + institutionTableName = "institutions2024" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2024} tsTableNameQ1 = "ts2024_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2024_Q1} tsTableNameQ2 = "ts2024_q2" @@ -111,6 +158,8 @@ hmda { 2023 { tsTableName = "transmittalsheet2023" tsTableName = ${?TS_TABLE_NAME_2023} + institutionTableName = "institutions2023" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2023} tsTableNameQ1 = "ts2023_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2023_Q1} tsTableNameQ2 = "ts2023_q2" @@ -129,6 +178,8 @@ hmda { 2022 { tsTableName = "transmittalsheet2022" tsTableName = ${?TS_TABLE_NAME_2022} + institutionTableName = "institutions2022" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2022} larTableName = "loanapplicationregister2022" larTableName = ${?LAR_TABLE_NAME_2022} tsTableNameQ1 = "ts2022_q1" @@ -147,6 +198,8 @@ hmda { 2021 { tsTableName = "transmittalsheet2021" tsTableName = ${?TS_TABLE_NAME_2021} + institutionTableName = "institutions2021" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2021} tsTableNameQ1 = "ts2021_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2021_Q1} tsTableNameQ2 = "ts2021_q2" @@ -165,6 +218,8 @@ hmda { 2020 { tsTableName = "transmittalsheet2020" tsTableName = ${?TS_TABLE_NAME_2020} + institutionTableName = "institutions2020" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2020} tsTableNameQ1 = "ts2020_q1" tsTableNameQ1 = ${?TS_TABLE_NAME_2020_Q1} tsTableNameQ2 = "ts2020_q2" @@ -185,6 +240,8 @@ hmda { 2019 { tsTableName = "transmittalsheet2019" tsTableName = ${?TS_TABLE_NAME_2019} + institutionTableName = "institutions2019" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2019} larTableName = "loanapplicationregister2019" larTableName = ${?LAR_TABLE_NAME_2019} historyTableName = "submission_history" @@ -193,6 +250,8 @@ hmda { 2018 { tsTableName = "transmittalsheet2018" tsTableName = ${?TS_TABLE_NAME_2018} + institutionTableName = "institutions2018" + institutionTableName = ${?INSTITUTION_TABLE_NAME_2018} larTableName = "loanapplicationregister2018" larTableName = ${?LAR_TABLE_NAME_2018} historyTableName = "submission_history" diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index b4d8bd216e..2bfd6fe1ab 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -3,6 +3,7 @@ package hmda.analytics import akka.Done import akka.actor.{ActorSystem, typed} import akka.actor.typed.scaladsl.adapter._ +import akka.cluster.sharding.typed.scaladsl.{ClusterSharding, EntityRef} import akka.kafka.scaladsl.Consumer.DrainingControl import akka.kafka.scaladsl.{Committer, Consumer} import akka.kafka.{CommitterSettings, ConsumerSettings, Subscriptions} @@ -12,12 +13,17 @@ import akka.util.{ByteString, Timeout} import com.typesafe.config.ConfigFactory import hmda.analytics.query._ import hmda.messages.HmdaMessageFilter +import hmda.messages.institution.InstitutionCommands.{GetInstitution, InstitutionCommand, InstitutionStop, ModifyInstitution} +import hmda.messages.institution.InstitutionEvents.InstitutionEvent import hmda.messages.pubsub.{HmdaGroups, HmdaTopics} import hmda.model.filing.lar.LoanApplicationRegister import hmda.model.filing.submission.SubmissionId import hmda.model.filing.ts.TransmittalSheet +import hmda.model.institution.{Agency, Institution, Respondent} import hmda.parser.filing.lar.LarCsvParser import hmda.parser.filing.ts.TsCsvParser +import hmda.persistence.institution.InstitutionPersistence +import hmda.persistence.institution.InstitutionPersistence.startShardRegion import hmda.publication.KafkaUtils._ import hmda.query.DbConfiguration.dbConfig import hmda.query.HmdaQuery.{readRawData, readSubmission} @@ -32,7 +38,7 @@ import scala.concurrent.{ExecutionContextExecutor, Future} import scala.concurrent.duration._ // $COVERAGE-OFF$ -object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarComponent with SubmissionHistoryComponent { +object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarComponent with SubmissionHistoryComponent with InstitutionTSComponent{ val log = LoggerFactory.getLogger("hmda") @@ -47,7 +53,8 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo | |___/ """.stripMargin) - implicit val system: ActorSystem = ActorSystem() + val config = ConfigFactory.load() + implicit val system: ActorSystem = ActorSystem(config.getString("hmda.cluster.name")) implicit val typedSystem: typed.ActorSystem[Nothing] = system.toTyped implicit val materializer: Materializer = Materializer(system) implicit val ec: ExecutionContextExecutor = system.dispatcher @@ -55,11 +62,15 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo implicit val timeout: Timeout = Timeout(5.seconds) val kafkaConfig = system.settings.config.getConfig("akka.kafka.consumer") - val config = ConfigFactory.load() val parallelism = config.getInt("hmda.analytics.parallelism") val larDeletion = config.getBoolean("hmda.analytics.larDeletion") val historyInsertion = config.getBoolean("hmda.analytics.historyInsertion") val tsDeletion = config.getBoolean("hmda.analytics.tsDeletion") + val sharding = ClusterSharding(typedSystem) + + startShardRegion(sharding) + + /** * Note: hmda-analytics microservice reads the JDBC_URL env var from inst-postgres-credentials secret. * In beta namespace this environment variable has currentSchema=hmda_beta_user appended to it to change the schema @@ -83,6 +94,18 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo } } + private def getInstitutionRepo(msg: String, year: String, quarter: String): InstitutionRepository = { + try { + val institutionTable = getTableName("hmda.analytics.%s.institutionTableName%s", year, quarter) + new InstitutionRepository(dbConfig, institutionTable) + } catch { + case e: Exception => + val errMsg = s"$msg: ${e.getMessage}" + log.error(errMsg) + throw new IllegalArgumentException(errMsg) + } + } + private def getTsRepo(msg: String, year: String, quarter: String): TransmittalSheetRepository = { try { val tsTable = getTableName("hmda.analytics.%s.tsTableName%s", year, quarter) @@ -185,24 +208,37 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo .filter(t => t.LEI != "" && t.institutionName != "") .map(ts => TransmittalSheetConverter(ts, submissionIdOption)) .mapAsync(1) { ts => - val repo = getTsRepo( + val tsRepo = getTsRepo( s"Unable to discern period from $submissionId to insert TS rows", submissionId.period.year.toString, submissionId.period.quarter.getOrElse("") ) + val institutionRepo = getInstitutionRepo( + s"Unable to discern period from $submissionId to update panel rows", + submissionId.period.year.toString, + submissionId.period.quarter.getOrElse("") + ) + val institutionEntity=generateInstitutionEntity(ts) + val institution= Institution.empty + val institutionPersistence = InstitutionPersistence.selectInstitution(sharding, institutionEntity.lei, institutionEntity.activityYear) + val enforceQuarterly = submissionId.period.quarter.isDefined + if (ts.year>2023) {institutionRepo.updateByLei(generateInstitutionEntity(ts))} for { + originalInstituion <- institutionPersistence ? GetInstitution + _ <- modifyActorPersistenceInstitution(institution, originalInstituion,ts,institutionPersistence) signdate <- signDate firstsigndate <- firstSignDateSubmissionHistory insertorupdate <- { - println("first sign date") - println(firstsigndate) val resolvedSignDate = Some(signdate.getOrElse(0L)) val resolvedFirstSignDate = { if (firstsigndate.head == 0) resolvedSignDate else Some(firstsigndate.head) } - repo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) + val tsRepoResult = tsRepo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) + institutionPersistence ! InstitutionStop + tsRepoResult + } } yield insertorupdate } @@ -310,5 +346,49 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo } + private def modifyActorPersistenceInstitution( + incomingInstitution: Institution, + originalInstOpt: Option[Institution], + transmittalSheetEntity: TransmittalSheetEntity, + institutionPersistence: EntityRef[InstitutionCommand] + ): Future[InstitutionEvent] = { + val originalFilerFlag = originalInstOpt.getOrElse(Institution.empty).hmdaFiler + val originalHasFiledQ1Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ1 + val originalHasFiledQ2Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ2 + val originalHasFiledQ3Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ3 + val institutionId_2017Flag =originalInstOpt.getOrElse(Institution.empty).institutionId_2017 + val emailDomainsFlag = originalInstOpt.getOrElse(Institution.empty).emailDomains + + val iFilerFlagsSet = incomingInstitution.copy( + LEI = transmittalSheetEntity.lei, + activityYear = transmittalSheetEntity.year, + agency =Agency.valueOf( transmittalSheetEntity.agency), + taxId = Some(transmittalSheetEntity.taxId), + respondent = Respondent(Some(transmittalSheetEntity.institutionName),Some(transmittalSheetEntity.state),Some(transmittalSheetEntity.city)), + institutionId_2017=institutionId_2017Flag, + emailDomains=emailDomainsFlag, + hmdaFiler = originalFilerFlag, + quarterlyFilerHasFiledQ1 = originalHasFiledQ1Flag, + quarterlyFilerHasFiledQ2 = originalHasFiledQ2Flag, + quarterlyFilerHasFiledQ3 = originalHasFiledQ3Flag + ) + + institutionPersistence ? (ref => ModifyInstitution(iFilerFlagsSet, ref)) + + + } + + private def generateInstitutionEntity(ts: TransmittalSheetEntity) = { + + InstitutionTSEntity( + lei = ts.lei, + activityYear = ts.year, + agency = ts.agency, + taxId = ts.taxId, + respondentName = ts.institutionName, + respondentState = ts.state, + respondentCity = ts.city, + ) + } } // $COVERAGE-ON$ diff --git a/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala new file mode 100644 index 0000000000..151ff566e1 --- /dev/null +++ b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala @@ -0,0 +1,93 @@ +package hmda.analytics.query + +import com.typesafe.config.{Config, ConfigFactory} +import hmda.query.DbConfiguration._ +import hmda.query.repository.TableRepository +import hmda.query.ts.{TransmittalSheetEntity, TransmittalSheetRepository} +import slick.basic.DatabaseConfig +import slick.jdbc.JdbcProfile + +import scala.concurrent.Future + +trait InstitutionTSComponent { + + import dbConfig.profile.api._ + + class InstitutionsTable(tag: Tag, tableName: String) extends Table[InstitutionTSEntity](tag, tableName) { + def lei = column[String]("lei", O.PrimaryKey) + def activityYear = column[Int]("activity_year") + def agency = column[Int]("agency") + def institutionType = column[Int]("institution_type") + def id2017 = column[String]("id2017") + def taxId = column[String]("tax_id") + def rssd = column[Int]("rssd") + def respondentName = column[String]("respondent_name") + def respondentState = column[String]("respondent_state") + def respondentCity = column[String]("respondent_city") + def parentIdRssd = column[Int]("parent_id_rssd") + def parentName = column[String]("parent_name") + def assets = column[Long]("assets") + def otherLenderCode = column[Int]("other_lender_code") + def topHolderIdRssd = column[Int]("topholder_id_rssd") + def topHolderName = column[String]("topholder_name") + def hmdaFiler = column[Boolean]("hmda_filer") + def quarterlyFiler = column[Boolean]("quarterly_filer") + def quarterlyFilerHasFiledQ1 = column[Boolean]("quarterly_filer_has_filed_q1") + def quarterlyFilerHasFiledQ2 = column[Boolean]("quarterly_filer_has_filed_q2") + def quarterlyFilerHasFiledQ3 = column[Boolean]("quarterly_filer_has_filed_q3") + def notes = column[String]("notes") + + def * = + ( + lei, + activityYear, + agency, + institutionType, + id2017, + taxId, + rssd, + respondentName, + respondentState, + respondentCity, + parentIdRssd, + parentName, + assets, + otherLenderCode, + topHolderIdRssd, + topHolderName, + hmdaFiler, + quarterlyFiler, + quarterlyFilerHasFiledQ1, + quarterlyFilerHasFiledQ2, + quarterlyFilerHasFiledQ3, + notes + ) <> (InstitutionTSEntity.tupled, InstitutionTSEntity.unapply) + } + + class InstitutionRepository(val config: DatabaseConfig[JdbcProfile], tableName: String) + extends TableRepository[InstitutionsTable, String] { + val table = TableQuery[InstitutionsTable]((tag: Tag) => new InstitutionsTable(tag, tableName)) + def getId(table: InstitutionsTable) = table.lei + def deleteById(lei: String) = db.run(filterById(lei).delete) + def getQuarterlyFilers(exclusions: Seq[String]) = db.run( + table.filter(_.quarterlyFiler === true) + .filterNot(_.lei inSet exclusions) + .map(i => (i.lei, i.respondentName, i.agency)) + .result + ) + + + def updateByLei(inst: InstitutionTSEntity): Future[Int] = + db.run(table.insertOrUpdate(inst)) + def createSchema() = db.run(table.schema.createIfNotExists) + def dropSchema() = db.run(table.schema.dropIfExists) + } + + val institutionConfig: Config = ConfigFactory.load().getConfig("hmda.analytics") + + val yearsAvailable: Seq[String] = institutionConfig.getString("yearsAvailable").split(",").toSeq + + val institutionRepositories: Map[String, InstitutionRepository] = (for { + year <- yearsAvailable + } yield year -> new InstitutionRepository(dbConfig, s"institutions$year")).toMap +} diff --git a/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSEntity.scala b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSEntity.scala new file mode 100644 index 0000000000..38f0a5b90e --- /dev/null +++ b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSEntity.scala @@ -0,0 +1,28 @@ +package hmda.analytics.query + +case class InstitutionTSEntity( + lei: String = "", + activityYear: Int = 0, + agency: Int = 9, + institutionType: Int = -1, + id2017: String = "", + taxId: String = "", + rssd: Int = -1, + respondentName: String = "", + respondentState: String = "", + respondentCity: String = "", + parentIdRssd: Int = -1, + parentName: String = "", + assets: Long = 0, + otherLenderCode: Int = -1, + topHolderIdRssd: Int = -1, + topHolderName: String = "", + hmdaFiler: Boolean = false, + quarterlyFiler: Boolean = false, + quarterlyFilerHasFiledQ1: Boolean = false, + quarterlyFilerHasFiledQ2: Boolean = false, + quarterlyFilerHasFiledQ3: Boolean = false, + notes: String = "" +) { + def isEmpty: Boolean = lei == "" +} diff --git a/hmda/src/main/scala/hmda/HmdaPlatform.scala b/hmda/src/main/scala/hmda/HmdaPlatform.scala index 5f99b60b73..727fc418da 100644 --- a/hmda/src/main/scala/hmda/HmdaPlatform.scala +++ b/hmda/src/main/scala/hmda/HmdaPlatform.scala @@ -75,8 +75,7 @@ object HmdaPlatform extends App { // TODO: Fix this as initializing it here is not a good idea, this should be initialized in HmdaPersistence and passed into HmdaValidationError val stringKafkaProducer = KafkaUtils.getStringKafkaProducer(system) - val institutionKafkaProducer = KafkaUtils.getInstitutionKafkaProducer(system) - + //Start Persistence classic.spawn(HmdaPersistence(), HmdaPersistence.name) diff --git a/hmda/src/main/scala/hmda/api/http/filing/submissions/SignHttpApi.scala b/hmda/src/main/scala/hmda/api/http/filing/submissions/SignHttpApi.scala index a21630289e..cf56bee915 100644 --- a/hmda/src/main/scala/hmda/api/http/filing/submissions/SignHttpApi.scala +++ b/hmda/src/main/scala/hmda/api/http/filing/submissions/SignHttpApi.scala @@ -87,8 +87,7 @@ private class SignHttpApi(log: Logger, sharding: ClusterSharding)(implicit t: Ti uri: Uri, username: String ): Route = { - log.info(s"inside signSubmission ${lei} and ${seqNr}") - val submissionId = SubmissionId(lei, Period(year, quarter), seqNr) + val submissionId = SubmissionId(lei, Period(year, quarter), seqNr) if (!signed) badRequest(submissionId, uri, "Illegal argument: signed = false") else { val hmdaValidationError = selectHmdaValidationError(sharding, submissionId)