From 71bd762a526173ed44113416f834956d468c3663 Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Tue, 30 Jun 2026 15:01:17 -0400 Subject: [PATCH 01/11] tying ts updates to institution table updates --- ...InstitutionCommandsProtobufConverter.scala | 5 +- .../InstitutionCommandsSerializer.scala | 6 +- .../src/main/resources/application.conf | 18 ++++ .../hmda/analytics/HmdaAnalyticsApp.scala | 38 +++++++- .../query/InstitutionTSComponent.scala | 93 +++++++++++++++++++ .../analytics/query/InstitutionTSEntity.scala | 28 ++++++ .../institution/InstitutionPersistence.scala | 3 +- 7 files changed, 180 insertions(+), 11 deletions(-) create mode 100644 hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala create mode 100644 hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSEntity.scala 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/hmda-analytics/src/main/resources/application.conf b/hmda-analytics/src/main/resources/application.conf index 8f6e5ee3b3..825de6411b 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -57,6 +57,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 +77,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 +97,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 +117,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 +137,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 +157,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 +177,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 +199,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 +209,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..493de728c2 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -32,7 +32,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") @@ -83,6 +83,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 +197,28 @@ 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 enforceQuarterly = submissionId.period.quarter.isDefined for { 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)) + tsRepo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) + institutionRepo.updateByLei(generateInstituionEntity(ts)) } } yield insertorupdate } @@ -310,5 +326,17 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo } + private def generateInstituionEntity(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..b476fbe68b --- /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.institution") + + 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/persistence/institution/InstitutionPersistence.scala b/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala index 8343bb887e..f61711b5fd 100644 --- a/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala +++ b/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala @@ -18,6 +18,7 @@ import hmda.model.institution.{Institution, InstitutionDetail} import hmda.persistence.HmdaTypedPersistentActor import hmda.publication.KafkaUtils._ import com.typesafe.config.ConfigFactory +import hmda.util.FieldNullifyUtility.nullifyInstitutionFields import scala.concurrent.Future @@ -152,7 +153,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 } From 9e3e5bf19b1051a21ed62851955a1d8f75ec338d Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Tue, 30 Jun 2026 15:27:38 -0400 Subject: [PATCH 02/11] tying ts updates to institution table updates --- .../src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index 493de728c2..0f72b552b0 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -208,6 +208,7 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo submissionId.period.quarter.getOrElse("") ) val enforceQuarterly = submissionId.period.quarter.isDefined + if (ts.year>2023) {institutionRepo.updateByLei(generateInstituionEntity(ts))} for { signdate <- signDate firstsigndate <- firstSignDateSubmissionHistory @@ -218,7 +219,6 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo else Some(firstsigndate.head) } tsRepo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) - institutionRepo.updateByLei(generateInstituionEntity(ts)) } } yield insertorupdate } From 2c36bfc2f3589e86d9542023748a0334a5458c1d Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Mon, 6 Jul 2026 23:53:04 -0400 Subject: [PATCH 03/11] correcting application conf reference to list of available years --- .../scala/hmda/analytics/query/InstitutionTSComponent.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala index b476fbe68b..151ff566e1 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/query/InstitutionTSComponent.scala @@ -83,7 +83,7 @@ trait InstitutionTSComponent { def dropSchema() = db.run(table.schema.dropIfExists) } - val institutionConfig: Config = ConfigFactory.load().getConfig("hmda.institution") + val institutionConfig: Config = ConfigFactory.load().getConfig("hmda.analytics") val yearsAvailable: Seq[String] = institutionConfig.getString("yearsAvailable").split(",").toSeq From 8e72377208f4bf080d15fa78f11c93ad8f428742 Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Wed, 8 Jul 2026 12:54:00 -0400 Subject: [PATCH 04/11] extracting cassandra persistance logic for institutions to the common package --- .../institution/InstitutionPersistence.scala | 34 +++++++++++++------ hmda/src/main/scala/hmda/HmdaPlatform.scala | 3 +- .../http/filing/submissions/SignHttpApi.scala | 3 +- 3 files changed, 26 insertions(+), 14 deletions(-) rename {hmda => common}/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala (86%) diff --git a/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala similarity index 86% rename from hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala rename to common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala index f61711b5fd..59144ad9b7 100644 --- a/hmda/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala +++ b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala @@ -8,21 +8,33 @@ 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] { +object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionCommand, InstitutionEvent, InstitutionState] with App { override final val name = "Institution" override def behavior(entityId: String): Behavior[InstitutionCommand] = @@ -47,6 +59,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 { @@ -61,7 +74,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 { @@ -78,7 +91,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 { @@ -94,7 +107,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 { @@ -144,7 +157,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) @@ -170,7 +184,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 { 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) From 99b04891bc59b484232761d45931ddfd8f1c30ed Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Wed, 8 Jul 2026 12:59:59 -0400 Subject: [PATCH 05/11] resolving relocating of institution cassandra persistance object to common package --- .../hmda/persistence/institution/InstitutionPersistence.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala index 59144ad9b7..da031a95e4 100644 --- a/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala +++ b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala @@ -34,7 +34,7 @@ import org.apache.kafka.clients.producer.{KafkaProducer, Producer} import scala.concurrent.{ExecutionContext, Future} -object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionCommand, InstitutionEvent, InstitutionState] with App { +object InstitutionPersistence extends HmdaTypedPersistentActor[InstitutionCommand, InstitutionEvent, InstitutionState] { override final val name = "Institution" override def behavior(entityId: String): Behavior[InstitutionCommand] = From 48e9aad8733ec709f1b846e4bc818f0dad2f7037 Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Wed, 8 Jul 2026 14:24:11 -0400 Subject: [PATCH 06/11] working cassandra institution persistance into hmda analytics --- .../institution/InstitutionPersistence.scala | 2 + .../Bank0_clean_100_rows_2025_annual.txt | 2 +- .../src/main/resources/application.conf | 25 ++++++++-- .../hmda/analytics/HmdaAnalyticsApp.scala | 49 ++++++++++++++++++- 4 files changed, 72 insertions(+), 6 deletions(-) diff --git a/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala index da031a95e4..2177bb3821 100644 --- a/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala +++ b/common/src/main/scala/hmda/persistence/institution/InstitutionPersistence.scala @@ -40,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), @@ -198,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/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..51ca2297e5 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|Asata Bank 0|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 825de6411b..8f0cd0974d 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -6,9 +6,28 @@ # include "persistence-keyspace-sigv4.conf" 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 + } } cassandra-journal { diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index 0f72b552b0..b44610e3b4 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} @@ -10,14 +11,20 @@ import akka.stream.Materializer import akka.stream.scaladsl.{Keep, Sink, Source} import akka.util.{ByteString, Timeout} import com.typesafe.config.ConfigFactory +import hmda.analytics.HmdaAnalyticsApp.generateInstitutionEntity import hmda.analytics.query._ import hmda.messages.HmdaMessageFilter +import hmda.messages.institution.InstitutionCommands.{GetInstitution, InstitutionCommand, 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} @@ -60,6 +67,11 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo 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 @@ -207,9 +219,15 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo 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(generateInstituionEntity(ts))} + if (ts.year>2023) {institutionRepo.updateByLei(generateInstitutionEntity(ts))} for { + originalInstituion <- institutionPersistence ? GetInstitution + _ <- modifyInstitutionViaCassandra(institution, originalInstituion,ts,institutionPersistence) signdate <- signDate firstsigndate <- firstSignDateSubmissionHistory insertorupdate <- { @@ -326,7 +344,34 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo } - private def generateInstituionEntity(ts: TransmittalSheetEntity) = { + private def modifyInstitutionViaCassandra( + 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_2017 =originalInstOpt.getOrElse(Institution.empty).institutionId_2017 + + 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_2017, + hmdaFiler = originalFilerFlag, + quarterlyFilerHasFiledQ1 = originalHasFiledQ1Flag, + quarterlyFilerHasFiledQ2 = originalHasFiledQ2Flag, + quarterlyFilerHasFiledQ3 = originalHasFiledQ3Flag + ) + institutionPersistence ? (ref => ModifyInstitution(iFilerFlagsSet, ref)) + } + + private def generateInstitutionEntity(ts: TransmittalSheetEntity) = { InstitutionTSEntity( lei = ts.lei, From 1f531bdd6bc73a57cb77cc52eabbbe13d4b5ab32 Mon Sep 17 00:00:00 2001 From: lchen <73617864+lchen-2101@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:51:14 -0700 Subject: [PATCH 07/11] feat: enable cluster mode --- .../src/main/resources/application.conf | 22 +++++++++++++++++++ .../hmda/analytics/HmdaAnalyticsApp.scala | 5 ++--- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/hmda-analytics/src/main/resources/application.conf b/hmda-analytics/src/main/resources/application.conf index 8f0cd0974d..59a570d2ac 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -5,6 +5,8 @@ # AWS Keyspace with SigV4AuthProvider # include "persistence-keyspace-sigv4.conf" +hmda.cluster.name = hmdaAnalytics + akka { diagnostics { starvation-detector { @@ -28,6 +30,21 @@ akka { 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 { @@ -63,6 +80,11 @@ cassandra-snapshot-store { } hmda { + supervisor { + minBackOff = 10 + maxBackOff = 60 + randomFactor = 0.1 + } analytics { parallelism = 1 tsDeletion = true diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index b44610e3b4..bc45e70d9c 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -11,7 +11,6 @@ import akka.stream.Materializer import akka.stream.scaladsl.{Keep, Sink, Source} import akka.util.{ByteString, Timeout} import com.typesafe.config.ConfigFactory -import hmda.analytics.HmdaAnalyticsApp.generateInstitutionEntity import hmda.analytics.query._ import hmda.messages.HmdaMessageFilter import hmda.messages.institution.InstitutionCommands.{GetInstitution, InstitutionCommand, ModifyInstitution} @@ -54,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 @@ -62,7 +62,6 @@ 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") From d1d960d2b425e5f2cefc87b912056f3d5ce62a5a Mon Sep 17 00:00:00 2001 From: lchen <73617864+lchen-2101@users.noreply.github.com> Date: Wed, 8 Jul 2026 14:49:27 -0700 Subject: [PATCH 08/11] feat: refactor cassandra naming --- .../src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index bc45e70d9c..db52011508 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -225,8 +225,8 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo val enforceQuarterly = submissionId.period.quarter.isDefined if (ts.year>2023) {institutionRepo.updateByLei(generateInstitutionEntity(ts))} for { - originalInstituion <- institutionPersistence ? GetInstitution - _ <- modifyInstitutionViaCassandra(institution, originalInstituion,ts,institutionPersistence) + originalInstituion <- institutionPersistence ? GetInstitution + _ <- modifyActorPersistenceInstitution(institution, originalInstituion,ts,institutionPersistence) signdate <- signDate firstsigndate <- firstSignDateSubmissionHistory insertorupdate <- { @@ -343,7 +343,7 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo } - private def modifyInstitutionViaCassandra( + private def modifyActorPersistenceInstitution( incomingInstitution: Institution, originalInstOpt: Option[Institution], transmittalSheetEntity: TransmittalSheetEntity, From 522c5d825ca0aee3a6211df8a2c303f9c8bf9fce Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Wed, 8 Jul 2026 18:26:31 -0400 Subject: [PATCH 09/11] adding email domain to institution cassandra persist events --- .../scala/hmda/analytics/HmdaAnalyticsApp.scala | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index db52011508..331458372b 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -353,7 +353,13 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo val originalHasFiledQ1Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ1 val originalHasFiledQ2Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ2 val originalHasFiledQ3Flag = originalInstOpt.getOrElse(Institution.empty).quarterlyFilerHasFiledQ3 - val institutionId_2017 =originalInstOpt.getOrElse(Institution.empty).institutionId_2017 + val institutionId_2017Flag =originalInstOpt.getOrElse(Institution.empty).institutionId_2017 + val emailDomainsFlag = originalInstOpt.getOrElse(Institution.empty).emailDomains + + + log.info(s"Institution Name:" +transmittalSheetEntity.institutionName) + log.info(s"City Name:" +transmittalSheetEntity.city) + log.info(s"State Name:" +transmittalSheetEntity.state) val iFilerFlagsSet = incomingInstitution.copy( LEI = transmittalSheetEntity.lei, @@ -361,12 +367,16 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo agency =Agency.valueOf( transmittalSheetEntity.agency), taxId = Some(transmittalSheetEntity.taxId), respondent = Respondent(Some(transmittalSheetEntity.institutionName),Some(transmittalSheetEntity.state),Some(transmittalSheetEntity.city)), - institutionId_2017=institutionId_2017, + institutionId_2017=institutionId_2017Flag, + emailDomains=emailDomainsFlag, hmdaFiler = originalFilerFlag, quarterlyFilerHasFiledQ1 = originalHasFiledQ1Flag, quarterlyFilerHasFiledQ2 = originalHasFiledQ2Flag, quarterlyFilerHasFiledQ3 = originalHasFiledQ3Flag ) + log.info(s"Flag Institution Name:" +iFilerFlagsSet.respondent.name) + log.info(s"Flag City Name:" +iFilerFlagsSet.respondent.city) + log.info(s"Flag State Name:" +iFilerFlagsSet.respondent.state) institutionPersistence ? (ref => ModifyInstitution(iFilerFlagsSet, ref)) } From 1f8ae4bf236e48f18868ece33ebb170a44cd072a Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Thu, 9 Jul 2026 11:10:32 -0400 Subject: [PATCH 10/11] resolving issues with sharding not properly stopping after persist events --- .../bank0/Bank0_clean_100_rows_2025_annual.txt | 2 +- .../src/main/resources/application.conf | 14 ++++++++++++-- .../hmda/analytics/HmdaAnalyticsApp.scala | 18 ++++++++---------- 3 files changed, 21 insertions(+), 13 deletions(-) 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 51ca2297e5..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|Asata Bank 0|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 59a570d2ac..952ea6c3c1 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -38,10 +38,20 @@ akka { } cluster { - seed-nodes = [ - "akka://"${hmda.cluster.name}"@127.0.0.1:2551" + ] + cluster { + seed-nodes = [ + "akka://"${hmda.cluster.name}"@127.0.0.1:2551" + failure-detector.acceptable-heartbeat-pause = 12 + metrics.enabled = off + sharding { + number-of-shards = 500 + passivation.default-idle-strategy.idle-entity.timeout = 35 minutes + passivation.default-idle-strategy.idle-entity.timeout = ${?AKKA_CLUSTER_IDLE_ENTITY_TIMEOUT} + } + } downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider" min-nr-of-members = 1 } diff --git a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala index 331458372b..2bfd6fe1ab 100644 --- a/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala +++ b/hmda-analytics/src/main/scala/hmda/analytics/HmdaAnalyticsApp.scala @@ -13,7 +13,7 @@ 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, ModifyInstitution} +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 @@ -235,7 +235,10 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo if (firstsigndate.head == 0) resolvedSignDate else Some(firstsigndate.head) } - tsRepo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) + val tsRepoResult = tsRepo.insert(copyTs(ts, resolvedSignDate, resolvedFirstSignDate, enforceQuarterly)) + institutionPersistence ! InstitutionStop + tsRepoResult + } } yield insertorupdate } @@ -356,11 +359,6 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo val institutionId_2017Flag =originalInstOpt.getOrElse(Institution.empty).institutionId_2017 val emailDomainsFlag = originalInstOpt.getOrElse(Institution.empty).emailDomains - - log.info(s"Institution Name:" +transmittalSheetEntity.institutionName) - log.info(s"City Name:" +transmittalSheetEntity.city) - log.info(s"State Name:" +transmittalSheetEntity.state) - val iFilerFlagsSet = incomingInstitution.copy( LEI = transmittalSheetEntity.lei, activityYear = transmittalSheetEntity.year, @@ -374,10 +372,10 @@ object HmdaAnalyticsApp extends App with TransmittalSheetComponent with LarCompo quarterlyFilerHasFiledQ2 = originalHasFiledQ2Flag, quarterlyFilerHasFiledQ3 = originalHasFiledQ3Flag ) - log.info(s"Flag Institution Name:" +iFilerFlagsSet.respondent.name) - log.info(s"Flag City Name:" +iFilerFlagsSet.respondent.city) - log.info(s"Flag State Name:" +iFilerFlagsSet.respondent.state) + institutionPersistence ? (ref => ModifyInstitution(iFilerFlagsSet, ref)) + + } private def generateInstitutionEntity(ts: TransmittalSheetEntity) = { From cda7120d8258967d0396b742872b14bde8f3130d Mon Sep 17 00:00:00 2001 From: PatrickGoRaft Date: Thu, 9 Jul 2026 12:00:12 -0400 Subject: [PATCH 11/11] resolving issues with sharding not properly stopping after persist events --- hmda-analytics/src/main/resources/application.conf | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/hmda-analytics/src/main/resources/application.conf b/hmda-analytics/src/main/resources/application.conf index 952ea6c3c1..59a570d2ac 100644 --- a/hmda-analytics/src/main/resources/application.conf +++ b/hmda-analytics/src/main/resources/application.conf @@ -38,20 +38,10 @@ akka { } cluster { - + seed-nodes = [ + "akka://"${hmda.cluster.name}"@127.0.0.1:2551" ] - cluster { - seed-nodes = [ - "akka://"${hmda.cluster.name}"@127.0.0.1:2551" - failure-detector.acceptable-heartbeat-pause = 12 - metrics.enabled = off - sharding { - number-of-shards = 500 - passivation.default-idle-strategy.idle-entity.timeout = 35 minutes - passivation.default-idle-strategy.idle-entity.timeout = ${?AKKA_CLUSTER_IDLE_ENTITY_TIMEOUT} - } - } downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider" min-nr-of-members = 1 }