tying ts updates to institution table updates

hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala → common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala

@@ -8,25 +8,39 @@ 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"
 
   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))
+
 }

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(

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 =>

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

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"