Merge branch 'master' into cluster-management

Author: jmarin

README.md

@@ -144,6 +144,10 @@ export ZOOKEEPER_HOST=192.168.99.100
 export ZOOKEEPER_PORT=2181
 ```
 
+Alternatively, these dependencies (`Cassandra`, `Zookeeper` and `PostgreSQL`) can be started from `docker` providing default resources for the `HMDA Platform`:
+
+`docker-compose -f docker-dev.yml up`
+
 * If you want to use the sample files in this repo for testing the app, run the edits in demo mode. Otherwise, edit S025 will trigger for all files.
 
 ```shell

api/src/main/resources/application.conf

@@ -77,4 +77,21 @@ hmda {
   zookeeperPort = ${?ZOOKEEPER_PORT}
 }
 
+api-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    # Min number of threads to cap factor-based parallelism number to
+    parallelism-min = 2
+    # Parallelism (threads) ... ceil(available processors * factor)
+    parallelism-factor = 2.0
+    # Max number of threads to cap factor-based parallelism number to
+    parallelism-max = 10
+  }
+  # Throughput defines the maximum number of messages to be
+  # processed per actor before the thread jumps to the next actor.
+  # Set to 1 for as fair as possible.
+  throughput = 100
+}
+
 

api/src/main/scala/hmda/api/HmdaPlatform.scala

@@ -38,7 +38,7 @@ object HmdaPlatform {
     ClusterHttpManagement(cluster).start()
     val supervisor = createSupervisor(system)
     val querySupervisor = createQuerySupervisor(system)
-    implicit val ec = system.dispatcher
+    implicit val ec = system.dispatchers.lookup("api-dispatcher")
 
     startActors(system, supervisor, querySupervisor)
     startApi(system, querySupervisor)

docker-dev.yml

@@ -0,0 +1,23 @@
+version: '2'
+
+services:
+  zookeeper:
+    image: jplock/zookeeper
+    ports:
+      - '2181:2181'
+
+  cassandra:
+    image: cassandra
+    ports:
+      - '9042:9042'
+      - '7000:7000'
+      - '7199:7199'
+
+  query_db:
+    image: postgres:9.6.1
+    ports:
+      - '54321:5432'
+    environment:
+      POSTGRES_DB: hmda
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres

persistence-model/src/main/scala/hmda/persistence/model/HmdaActor.scala

@@ -6,6 +6,7 @@ trait HmdaActor extends Actor with ActorLogging {
 
   override def preStart(): Unit = {
     log.info(s"Actor started at ${self.path}")
+    log.info("Thread name for actor: " + Thread.currentThread().getName)
   }
 
   override def postStop(): Unit = {

persistence/src/main/resources/application.conf

@@ -1,3 +1,20 @@
 hmda {
   actor-flow-parallelism = 4
 }
+
+persistence-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    # Min number of threads to cap factor-based parallelism number to
+    parallelism-min = 2
+    # Parallelism (threads) ... ceil(available processors * factor)
+    parallelism-factor = 2.0
+    # Max number of threads to cap factor-based parallelism number to
+    parallelism-max = 10
+  }
+  # Throughput defines the maximum number of messages to be
+  # processed per actor before the thread jumps to the next actor.
+  # Set to 1 for as fair as possible.
+  throughput = 100
+}

persistence/src/main/scala/hmda/persistence/HmdaSupervisor.scala

@@ -16,7 +16,7 @@ object HmdaSupervisor {
   def props(): Props = Props(new HmdaSupervisor)
 
   def createSupervisor(system: ActorSystem): ActorRef = {
-    system.actorOf(HmdaSupervisor.props(), "supervisor")
+    system.actorOf(HmdaSupervisor.props().withDispatcher("persistence-dispatcher"), "supervisor")
   }
 }
 
@@ -54,47 +54,52 @@ class HmdaSupervisor extends HmdaSupervisorActor {
 
   override def createActor(name: String): ActorRef = name match {
     case id @ SingleLarValidation.name =>
-      val actor = context.actorOf(SingleLarValidation.props, id)
+      val actor = context.actorOf(SingleLarValidation.props.withDispatcher("persistence-dispatcher"), id)
       supervise(actor, id)
     case id @ InstitutionPersistence.name =>
-      val actor = context.actorOf(InstitutionPersistence.props, id)
+      val actor = context.actorOf(InstitutionPersistence.props.withDispatcher("persistence-dispatcher"), id)
       supervise(actor, id)
 
   }
 
   private def createHmdaFiling(filingPeriod: String): ActorRef = {
-    val actor = context.actorOf(HmdaFiling.props(filingPeriod), s"${HmdaFiling.name}-$filingPeriod")
+    val actor = context.actorOf(
+      HmdaFiling.props(filingPeriod).withDispatcher("persistence-dispatcher"),
+      s"${HmdaFiling.name}-$filingPeriod"
+    )
     supervise(actor, HmdaFiling.name)
   }
 
   private def createFilings(name: String, id: String): ActorRef = {
     val filingsId = s"$name-$id"
-    val actor = context.actorOf(FilingPersistence.props(id), filingsId)
+    val actor = context.actorOf(FilingPersistence.props(id).withDispatcher("persistence-dispatcher"), filingsId)
     supervise(actor, filingsId)
   }
 
   private def createSubmissions(name: String, institutionId: String, period: String): ActorRef = {
     val sId = s"$name-$institutionId-$period"
-    val actor = context.actorOf(SubmissionPersistence.props(institutionId, period), sId)
+    val actor = context.actorOf(SubmissionPersistence
+      .props(institutionId, period)
+      .withDispatcher("persistence-dispatcher"), sId)
     supervise(actor, sId)
   }
 
   private def createProcessingActor(name: String, submissionId: SubmissionId): ActorRef = name match {
     case id @ HmdaRawFile.name =>
       val actorId = s"$id-${submissionId.toString}"
-      val actor = context.actorOf(HmdaRawFile.props(submissionId), actorId)
+      val actor = context.actorOf(HmdaRawFile.props(submissionId).withDispatcher("persistence-dispatcher"), actorId)
       supervise(actor, actorId)
     case id @ HmdaFileParser.name =>
       val actorId = s"$id-${submissionId.toString}"
-      val actor = context.actorOf(HmdaFileParser.props(submissionId), actorId)
+      val actor = context.actorOf(HmdaFileParser.props(submissionId).withDispatcher("persistence-dispatcher"), actorId)
       supervise(actor, actorId)
     case id @ HmdaFileValidator.name =>
       val actorId = s"$id-${submissionId.toString}"
-      val actor = context.actorOf(HmdaFileValidator.props(submissionId), actorId)
+      val actor = context.actorOf(HmdaFileValidator.props(submissionId).withDispatcher("persistence-dispatcher"), actorId)
       supervise(actor, actorId)
     case id @ SubmissionManager.name =>
       val actorId = s"$id-${submissionId.toString}"
-      val actor = context.actorOf(SubmissionManager.props(submissionId), actorId)
+      val actor = context.actorOf(SubmissionManager.props(submissionId).withDispatcher("persistence-dispatcher"), actorId)
       supervise(actor, actorId)
   }
 

persistence/src/main/scala/hmda/persistence/institutions/FilingPersistence.scala

@@ -18,7 +18,7 @@ object FilingPersistence {
   def props(institutionId: String): Props = Props(new FilingPersistence(institutionId))
 
   def createFilings(institutionId: String, system: ActorSystem): ActorRef = {
-    system.actorOf(FilingPersistence.props(institutionId))
+    system.actorOf(FilingPersistence.props(institutionId).withDispatcher("persistence-dispatcher"))
   }
 
   case class FilingState(filings: Seq[Filing] = Nil) {

persistence/src/main/scala/hmda/persistence/institutions/InstitutionPersistence.scala

@@ -18,7 +18,7 @@ object InstitutionPersistence {
   def props: Props = Props(new InstitutionPersistence)
 
   def createInstitutions(system: ActorSystem): ActorRef = {
-    system.actorOf(InstitutionPersistence.props, "institutions")
+    system.actorOf(InstitutionPersistence.props.withDispatcher("persistence-dispatcher"), "institutions")
   }
 
   case class InstitutionPersistenceState(institutions: Set[Institution] = Set.empty[Institution]) {

persistence/src/main/scala/hmda/persistence/institutions/SubmissionPersistence.scala

@@ -19,7 +19,7 @@ object SubmissionPersistence {
   def props(institutionId: String, period: String): Props = Props(new SubmissionPersistence(institutionId, period))
 
   def createSubmissions(institutionId: String, period: String, system: ActorSystem): ActorRef = {
-    system.actorOf(SubmissionPersistence.props(institutionId, period))
+    system.actorOf(SubmissionPersistence.props(institutionId, period).withDispatcher("persistence-dispatcher"))
   }
 
   case class SubmissionState(submissions: Seq[Submission] = Nil) {