Merge pull request #5014 from lchen-2101/chore/cve_update_250424

chore: cve updates

Author: PatrickGoRaft

.github/workflows/helper-functions.yaml

@@ -8,6 +8,13 @@ on:
 jobs:
   test-service:
     runs-on: ubuntu-latest
+    services:
+      cassandra:
+        image: "cassandra:3.11.4"
+        ports:
+          - "9042:9042"
+    env:
+      HMDA_RUNTIME_MODE: "docker-compose"
 
     steps:
       - name: Checkout
@@ -19,7 +26,7 @@ jobs:
       - name: Setup JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'adopt'
 
       - name: Run hmda-platform Tests

README.md

@@ -174,15 +174,26 @@ export JAVA_HOME=$HOME/.asdf/installs/java/openjdk-13.0.2
 
 ### Running with sbt
 
-The HMDA Platform can run locally using [`sbt`](https://www.scala-sbt.org/) with an [embedded Cassandra](https://doc.akka.io/docs/alpakka-kafka/current/) and [embedded Kafka](https://doc.akka.io/docs/alpakka-kafka/current/). To get started:
+~~The HMDA Platform can run locally using [`sbt`](https://www.scala-sbt.org/) with an [embedded Cassandra](https://doc.akka.io/docs/alpakka-kafka/current/) and [embedded Kafka](https://doc.akka.io/docs/alpakka-kafka/current/). To get started:~~  
+_removing embedded cassandra allows us to use more up-to-date java versions_
 
+For local development, supporting services need to be started first in docker:
+
+1. Open terminal with `hmda-platform` root as the  working directory
+2. Start the supporting services of Kafka, Cassandra, and PostgreSQL:
+   ```bash
+   docker compose up -d kafka cassandra pg
+   ```
+   * the `-d` option runs the services in detached mode, so we can use the same terminal to the run remaining commands, to stop the services run:
+     ```bash
+     docker compose stop
+     ```
 1. Export the following environment variables:
     ```bash
     export CASSANDRA_CLUSTER_HOSTS=localhost
     export APP_PORT=2551
     ```
-2. Open terminal with `hmda-platform` root as the  working directory
-3. Start sbt and run the platform with the following commands:
+1. Start sbt and run the platform with the following commands:
     ```bash
     sbt
     [...]

build.sbt

@@ -3,7 +3,7 @@ import BuildSettings._
 import sbtassembly.AssemblyPlugin.autoImport.assemblyMergeStrategy
 import com.typesafe.sbt.packager.docker._
 
-lazy val commonDeps = Seq(logback, scalaTest, scalaCheck, akkaHttpSprayJson, testContainers, apacheCommonsIO, log4jToSlf4j, kubernetesApi)
+lazy val commonDeps = Seq(logback, scalaTest, scalaCheck, akkaHttpSprayJson, testContainers, apacheCommonsIO, log4jToSlf4j, kubernetesApi, scalaLogging)
 
 lazy val sparkDeps =
   Seq(
@@ -33,7 +33,6 @@ lazy val akkaDeps = Seq(
   akkaCors,
   mskdriver,
   akkaKafkaStreams,
-  embeddedKafka,
   alpakkaS3,
   akkaQuartzScheduler,
   alpakkaFile
@@ -47,8 +46,7 @@ lazy val akkaPersistenceDeps =
     akkaPersistenceQuery,
     akkaClusterShardingTyped,
     akkaPersistenceCassandra,
-    keyspacedriver,
-    cassandraLauncher
+    keyspacedriver
   )
 
 lazy val akkaHttpDeps =

common/src/main/scala/hmda/persistence/util/CassandraUtil.scala

@@ -13,37 +13,43 @@ class RealTimeConfig(val cmName: String, val ns: String) {
   private var currentConfig: Option[Config] = None
   private var factory: Option[SharedInformerFactory] = None
 
-  try {
-    val client = io.kubernetes.client.util.Config.defaultClient()
-    val api = new CoreV1Api(client)
-    factory = Option(new SharedInformerFactory(client))
-    val informer = factory.get.sharedIndexInformerFor((params: CallGeneratorParams) => {
-      api.listNamespacedConfigMapCall(
-        ns, null, null, null, s"metadata.name=$cmName", null, null, params.resourceVersion, null, null, params.timeoutSeconds, params.watch, null)
-    }, classOf[V1ConfigMap], classOf[V1ConfigMapList])
-    informer.addEventHandler(new ResourceEventHandler[V1ConfigMap] {
-      override def onAdd(obj: V1ConfigMap): Unit = {
-        log.debug("cm added: {}", obj)
-        setConfig(obj)
-      }
-
-      override def onUpdate(oldObj: V1ConfigMap, newObj: V1ConfigMap): Unit = {
-        log.debug("cm updated: {}", newObj)
-        setConfig(newObj)
-      }
-
-      override def onDelete(obj: V1ConfigMap, deletedFinalStateUnknown: Boolean): Unit = log.warn("cm deleted: {}, deleteStateUnknown: {}", obj, deletedFinalStateUnknown)
-    })
-
-    factory.get.startAllRegisteredInformers()
-    setConfig(api.readNamespacedConfigMap(cmName, ns, null))
-  } catch {
-    case e: ApiException =>
-      log.error(s"Failed to setup informer, most likely role permission issues. ${e.getResponseBody}", e)
-      factory.get.stopAllRegisteredInformers()
-    case e: Throwable =>
-      log.error(s"Failed to setup informer", e)
-      factory.get.stopAllRegisteredInformers()
+  private val config = ConfigFactory.load()
+
+  private val runMode = if (config.hasPath("hmda.runtime.mode")) config.getString("hmda.runtime.mode") else "dev"
+
+  if (runMode == "kubernetes") {
+    try {
+      val client = io.kubernetes.client.util.Config.defaultClient()
+      val api = new CoreV1Api(client)
+      factory = Option(new SharedInformerFactory(client))
+      val informer = factory.get.sharedIndexInformerFor((params: CallGeneratorParams) => {
+        api.listNamespacedConfigMapCall(
+          ns, null, null, null, s"metadata.name=$cmName", null, null, params.resourceVersion, null, null, params.timeoutSeconds, params.watch, null)
+      }, classOf[V1ConfigMap], classOf[V1ConfigMapList])
+      informer.addEventHandler(new ResourceEventHandler[V1ConfigMap] {
+        override def onAdd(obj: V1ConfigMap): Unit = {
+          log.debug("cm added: {}", obj)
+          setConfig(obj)
+        }
+
+        override def onUpdate(oldObj: V1ConfigMap, newObj: V1ConfigMap): Unit = {
+          log.debug("cm updated: {}", newObj)
+          setConfig(newObj)
+        }
+
+        override def onDelete(obj: V1ConfigMap, deletedFinalStateUnknown: Boolean): Unit = log.warn("cm deleted: {}, deleteStateUnknown: {}", obj, deletedFinalStateUnknown)
+      })
+
+      factory.get.startAllRegisteredInformers()
+      setConfig(api.readNamespacedConfigMap(cmName, ns, null))
+    } catch {
+      case e: ApiException =>
+        log.error(s"Failed to setup informer, most likely role permission issues. ${e.getResponseBody}", e)
+        factory.get.stopAllRegisteredInformers()
+      case e: Throwable =>
+        log.error(s"Failed to setup informer", e)
+        factory.get.stopAllRegisteredInformers()
+    }
   }
 
   private def setConfig(cm: V1ConfigMap): Unit = {

common/src/main/scala/hmda/util/RealTimeConfig.scala

@@ -2,18 +2,16 @@ package hmda.persistence
 
 import java.time.Instant
 import java.util.concurrent.TimeUnit
-
 import akka.actor
 import akka.actor.testkit.typed.scaladsl.TestProbe
 import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.scaladsl.adapter._
-import akka.actor.typed.{ ActorRef, ActorSystem, Behavior, TypedActorContext }
+import akka.actor.typed.{ActorRef, ActorSystem, Behavior, TypedActorContext}
 import akka.persistence.typed.PersistenceId
 import akka.persistence.typed.scaladsl.EventSourcedBehavior.CommandHandler
-import akka.persistence.typed.scaladsl.{ Effect, EventSourcedBehavior }
-import hmda.persistence.util.CassandraUtil
+import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior}
 import org.scalacheck.Gen
-import org.scalatest.{ BeforeAndAfterAll, WordSpec }
+import org.scalatest.{BeforeAndAfterAll, WordSpec}
 
 import scala.concurrent.duration._
 
@@ -28,13 +26,11 @@ abstract class AkkaCassandraPersistenceSpec extends WordSpec with BeforeAndAfter
   implicit val typedSystem: ActorSystem[_]
 
   override def beforeAll(): Unit = {
-    CassandraUtil.startEmbeddedCassandra()
     awaitPersistenceInit()
     super.beforeAll()
   }
 
   override def afterAll(): Unit = {
-    CassandraUtil.shutdown()
     system.terminate()
     super.afterAll()
   }

common/src/test/scala/hmda/persistence/AkkaCassandraPersistenceSpec.scala

@@ -1,8 +1,8 @@
 package hmda.dataBrowser.services
 
 import akka.NotUsed
+import akka.actor.typed.ActorSystem
 import akka.http.scaladsl.model.ContentTypes
-import akka.stream.Materializer
 import akka.stream.alpakka.s3.S3Headers
 import akka.stream.alpakka.s3.scaladsl.S3
 import akka.stream.scaladsl.{ Sink, Source }
@@ -16,7 +16,7 @@ import org.slf4j.LoggerFactory
 
 // $COVERAGE-OFF$
 // All this does is use the Alpakka S3 APIs
-class S3FileService(implicit mat: Materializer) extends FileService with Settings {
+class S3FileService(implicit system: ActorSystem[Nothing]) extends FileService with Settings {
 
   private final val log = LoggerFactory.getLogger(getClass)
 

common/src/test/scala/hmda/utils/EmbeddedCassandra.scala

@@ -12,10 +12,8 @@ import akka.stream.Materializer
 import com.typesafe.config.ConfigFactory
 import hmda.api.HmdaApi
 import hmda.persistence.HmdaPersistence
-import hmda.persistence.util.CassandraUtil
 import hmda.publication.{HmdaPublication, KafkaUtils}
 import hmda.validation.HmdaValidation
-import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
 import org.slf4j.LoggerFactory
 
 // $COVERAGE-OFF$
@@ -72,21 +70,8 @@ object HmdaPlatform extends App {
   implicit val mat     = Materializer(system)
   implicit val cluster = Cluster(system)
 
-  if (runtimeMode == "dcos" || runtimeMode == "kubernetes" || runtimeMode == "docker-compose" || runtimeMode == "kind") {
-    ClusterBootstrap(system).start()
-    AkkaManagement(system).start()
-  }
-
-  if (runtimeMode == "dev") {
-    CassandraUtil.startEmbeddedCassandra()
-    AkkaManagement(system).start()
-    implicit val embeddedKafkaConfig: EmbeddedKafkaConfig = EmbeddedKafkaConfig(
-      sys.env.getOrElse("HMDA_LOCAL_KAFKA_PORT", "9092").toInt,
-      sys.env.getOrElse("HMDA_LOCAL_ZK_PORT", "2182").toInt,
-      Map("offsets.topic.replication.factor" -> "1", "zookeeper.connection.timeout.ms" -> "20000")
-    )
-    EmbeddedKafka.start()
-  }
+  ClusterBootstrap(system).start()
+  AkkaManagement(system).start()
 
   // 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)