Merge pull request #4975 from PatrickGoRaft/dev-keyspace-update

Dev keyspace update
cfpb · Jan 31, 2025 · fa2a78a · fa2a78a
2 parents b266cec + 042d87b
commit fa2a78a
Show file tree

Hide file tree

Showing 16 changed files with 401 additions and 19 deletions.
diff --git a/build.sbt b/build.sbt
@@ -47,6 +47,7 @@ lazy val akkaPersistenceDeps =
     akkaPersistenceQuery,
     akkaClusterShardingTyped,
     akkaPersistenceCassandra,
+    keyspacedriver,
     cassandraLauncher
   )
 
@@ -180,7 +181,7 @@ lazy val `hmda-platform` = (project in file("hmda"))
           val oldStrategy = (assembly / assemblyMergeStrategy).value
           oldStrategy(x)
       },
-      reStart / envVars ++= Map("CASSANDRA_CLUSTER_HOSTS" -> "localhost", "APP_PORT" -> "2551"),
+     reStart / envVars ++= Map("CASSANDRA_CLUSTER_HOSTS" -> "localhost", "APP_PORT" -> "2551"),
     ),
     dockerSettings,
     packageSettings

diff --git a/common/src/main/resources/persistence-keyspace-sigv4.conf b/common/src/main/resources/persistence-keyspace-sigv4.conf
@@ -0,0 +1,108 @@
+include "serialization.conf"
+
+akka {
+
+  persistence {
+    journal.plugin = "akka.persistence.cassandra.journal"
+    snapshot-store.plugin = "akka.persistence.cassandra.snapshot"
+
+    query {
+      journal.id = "akka.persistence.cassandra.query"
+    }
+
+    cassandra {
+      events-by-tag {
+        max-message-batch-size = 30
+        max-message-batch-size = ${?TAG_BATCH_SIZE}
+        flush-interval = 150ms
+        flush-interval = ${?TAG_FLUSH_INTERVAL}
+      }
+      journal {
+        keyspace = "hmda2_journal"
+        keyspace = ${?CASSANDRA_JOURNAL_KEYSPACE}
+        table = "journal"
+        events-by-tag.max-message-batch-size = 30
+        events-by-tag.max-message-batch-size = ${?TAG_BATCH_SIZE}
+        events-by-tag.flush-interval = 150ms
+        events-by-tag.flush-interval = ${?TAG_FLUSH_INTERVAL}
+      }
+      query {
+        # reference: https://github.com/akka/akka-persistence-cassandra/blob/v0.101/core/src/main/resources/reference.conf
+        # Sequence numbers for a persistenceId is assumed to be monotonically increasing
+        # without gaps. That is used for detecting missing events.
+        # In early versions of the journal that might not be true and therefore
+        # this can be relaxed by setting this property to off.
+        gap-free-sequence-numbers = off
+      }
+      snapshot {
+        keyspace = "hmda2_snapshot"
+        keyspace = ${?CASSANDRA_SNAPSHOT_KEYSPACE}
+        table = "snapshot"
+      }
+    }
+  }
+
+
+}
+
+datastax-java-driver {
+  profiles {
+      akka-persistence-cassandra-profile {
+           basic {
+             request.consistency = LOCAL_QUORUM
+           }
+      }
+  }
+  basic {
+      contact-points = ["localhost:9042"]
+      contact-points = [${?CASSANDRA_CLUSTER_HOSTS}]
+      load-balancing-policy.local-datacenter = ""
+      load-balancing-policy.local-datacenter = ${?CASSANDRA_CLUSTER_DC}
+      load-balancing-policy.slow-replica-avoidance = false 
+      request.consistency = LOCAL_QUORUM      
+  }
+
+
+
+  advanced {
+
+    advanced.protocol {
+      version = V4
+    }
+    auth-provider {
+      class = software.aws.mcs.auth.SigV4AuthProvider
+      aws-region = "us-east-1"
+    }
+    request-tracker {
+      classes = [RequestLogger]
+      logs {
+        slow {
+          threshold = 1 second
+          enabled = true
+        }
+      }
+    }
+    ssl-engine-factory {
+        class = DefaultSslEngineFactory
+        truststore-path = ${?CASSANDRA_TRUSTSTORE_FILE}
+        truststore-password = ${?CASSANDRA_TRUSTSTORE_PASSWORD}
+        hostname-validation = false
+      }    
+
+  }
+}
+
+kafka {
+  hosts = "localhost:9092"
+  hosts = ${?KAFKA_CLUSTER_HOSTS}
+  idle-timeout = 5
+  idle-timeout = ${?KAFKA_IDLE_TIMEOUT}
+  security.protocol=""
+  security.protocol=${?KAFKA_SECURITY}
+  ssl.truststore.location = ""
+  ssl.truststore.location = ${?TRUSTSTORE_PATH}
+  ssl.truststore.password = ""
+  ssl.truststore.password = ${?TRUSTSTORE_PASSWORD}
+  ssl.endpoint = ""
+  ssl.endpoint = ${?KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG}
+}
diff --git a/common/src/main/resources/persistence-keyspace.conf b/common/src/main/resources/persistence-keyspace.conf
@@ -0,0 +1,109 @@
+include "serialization.conf"
+
+akka {
+
+  persistence {
+    journal.plugin = "akka.persistence.cassandra.journal"
+    snapshot-store.plugin = "akka.persistence.cassandra.snapshot"
+
+    query {
+      journal.id = "akka.persistence.cassandra.query"
+    }
+
+    cassandra {
+      events-by-tag {
+        max-message-batch-size = 30
+        max-message-batch-size = ${?TAG_BATCH_SIZE}
+        flush-interval = 150ms
+        flush-interval = ${?TAG_FLUSH_INTERVAL}
+      }
+      journal {
+        keyspace = "hmda2_journal"
+        keyspace = ${?CASSANDRA_JOURNAL_KEYSPACE}
+        table = "journal"
+        events-by-tag.max-message-batch-size = 30
+        events-by-tag.max-message-batch-size = ${?TAG_BATCH_SIZE}
+        events-by-tag.flush-interval = 150ms
+        events-by-tag.flush-interval = ${?TAG_FLUSH_INTERVAL}
+      }
+      query {
+        # reference: https://github.com/akka/akka-persistence-cassandra/blob/v0.101/core/src/main/resources/reference.conf
+        # Sequence numbers for a persistenceId is assumed to be monotonically increasing
+        # without gaps. That is used for detecting missing events.
+        # In early versions of the journal that might not be true and therefore
+        # this can be relaxed by setting this property to off.
+        gap-free-sequence-numbers = off
+      }
+      snapshot {
+        keyspace = "hmda2_snapshot"
+        keyspace = ${?CASSANDRA_SNAPSHOT_KEYSPACE}
+        table = "snapshot"
+      }
+    }
+  }
+
+
+}
+
+datastax-java-driver {
+
+  profiles {
+    akka-persistence-cassandra-profile {
+      basic {
+        request.consistency = LOCAL_QUORUM
+      }
+    }
+  }
+
+  basic {
+    contact-points = ["localhost:9042"]
+    contact-points = [${?CASSANDRA_CLUSTER_HOSTS}]
+    load-balancing-policy.local-datacenter = ""
+    load-balancing-policy.local-datacenter = ${?CASSANDRA_CLUSTER_DC}
+    load-balancing-policy.slow-replica-avoidance = false
+    request.consistency = LOCAL_QUORUM
+  }
+
+  advanced {
+    advanced.protocol {
+      version = V4
+    }
+    auth-provider {
+      class = PlainTextAuthProvider
+      username = ""
+      username = ${?CASSANDRA_CLUSTER_USERNAME}
+      password = ""
+      password = ${?CASSANDRA_CLUSTER_PASSWORD}
+    }
+    request-tracker {
+      classes = [RequestLogger]
+      logs {
+        slow {
+          threshold = 1 second
+          enabled = true
+        }
+      }
+    }
+    ssl-engine-factory {
+      class = DefaultSslEngineFactory
+      truststore-path = ${?CASSANDRA_TRUSTSTORE_FILE}
+      truststore-password = ${?CASSANDRA_TRUSTSTORE_PASSWORD}
+      hostname-validation = false
+    }
+  }
+}
+
+kafka {
+  hosts = "localhost:9092"
+  hosts = ${?KAFKA_CLUSTER_HOSTS}
+  idle-timeout = 5
+  idle-timeout = ${?KAFKA_IDLE_TIMEOUT}
+  security.protocol=""
+  security.protocol=${?KAFKA_SECURITY}
+  ssl.truststore.location = ""
+  ssl.truststore.location = ${?TRUSTSTORE_PATH}
+  ssl.truststore.password = ""
+  ssl.truststore.password = ${?TRUSTSTORE_PASSWORD}
+  ssl.endpoint = ""
+  ssl.endpoint = ${?KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG}
+}
diff --git a/common/src/main/resources/persistence.conf b/common/src/main/resources/persistence.conf
@@ -16,6 +16,10 @@ akka {
         max-message-batch-size = ${?TAG_BATCH_SIZE}
         flush-interval = 150ms
         flush-interval = ${?TAG_FLUSH_INTERVAL}
+        # read-profile = "akka-persistence-cassandra-profile-keyspaces"
+        # write-profile = "akka-persistence-cassandra-profile-keyspaces"
+        read-profile1 = ${?CASSANDRA_PROFILE}
+        write-profile = ${?CASSANDRA_PROFILE}
       }
       journal {
         keyspace = "hmda2_journal"
@@ -46,12 +50,24 @@ akka {
 }
 
 datastax-java-driver {
+
+  profiles {
+    akka-persistence-cassandra-profile  {
+    #  akka-persistence-cassandra-profile-keyspaces {
+      basic {
+        request.consistency = LOCAL_QUORUM
+      }
+    }
+  }
+
   basic {
       contact-points = ["localhost:9042"]
-      contact-points = [${?CASSANDRA_CLUSTER_HOSTS}":9042"]
+      contact-points = [${?CASSANDRA_CLUSTER_HOSTS}]
       load-balancing-policy.local-datacenter = "datacenter1"
       load-balancing-policy.local-datacenter = ${?CASSANDRA_CLUSTER_DC}
+      load-balancing-policy.slow-replica-avoidance = false
   }
+
   advanced {
     advanced.protocol {
       version = V4
@@ -64,7 +80,7 @@ datastax-java-driver {
       password = ${?CASSANDRA_CLUSTER_PASSWORD}
     }
     request-tracker {
-      class = RequestLogger
+      classes = [RequestLogger]
       logs {
         slow {
           threshold = 1 second
@@ -82,17 +98,17 @@ kafka {
   idle-timeout = ${?KAFKA_IDLE_TIMEOUT}
   security.protocol=""
   security.protocol=${?KAFKA_SECURITY}
-  // ssl.truststore.location = ""
-  // ssl.truststore.location = ${?TRUSTSTORE_PATH}
-  // ssl.truststore.password = ""
-  // ssl.truststore.password = ${?TRUSTSTORE_PASSWORD}
-  // ssl.endpoint = ""
-  // ssl.endpoint = ${?KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG}
+
+ # ssl.truststore.location = ""
+ # ssl.truststore.location = ${?TRUSTSTORE_PATH}
+ # ssl.truststore.password = ""
+ # ssl.truststore.password = ${?TRUSTSTORE_PASSWORD}
+ # ssl.endpoint = ""
+ # ssl.endpoint = ${?KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG}
   sasl.mechanism="AWS_MSK_IAM"
-  //sasl.mechanism=${?KAFKA_SASL_MECHANISM}
+ #sasl.mechanism=${?KAFKA_SASL_MECHANISM}
   sasl.jaas.config="software.amazon.msk.auth.iam.IAMLoginModule required;"
-  //sasl.jaas.config="{?KAFKA_SASL_JAAS_CONFIG}"
+  # sasl.jaas.config="{?KAFKA_SASL_JAAS_CONFIG}"
   sasl.client.callback.handler.class="software.amazon.msk.auth.iam.IAMClientCallbackHandler"
-  //sasl.client.callback.handler.class="{?KAFKA_SASL_CLASS}"
-}
-
+  #sasl.client.callback.handler.class="{?KAFKA_SASL_CLASS}"
+}
diff --git a/hmda-analytics/src/main/resources/application.conf b/hmda-analytics/src/main/resources/application.conf
@@ -1,4 +1,9 @@
-include "persistence.conf"
+# Apache Cassandra with PlainTextAuthProvider
+ include "persistence.conf"
+# AWS Keyspace with PlainTextAuthProvider
+# include "persistence-keyspace.conf"
+# AWS Keyspace with SigV4AuthProvider
+# include "persistence-keyspace-sigv4.conf"
 
 akka {
   loglevel = info

diff --git a/hmda/src/main/resources/application-kubernetes.conf b/hmda/src/main/resources/application-kubernetes.conf
@@ -1,4 +1,9 @@
+# Apache Cassandra with PlainTextAuthProvider
 include "persistence.conf"
+# AWS Keyspace with PlainTextAuthProvider
+# include "persistence-keyspace.conf"
+# AWS Keyspace with SigV4AuthProvider
+# include "persistence-keyspace-sigv4.conf"
 include "hmda.conf"
 include "cors.conf"
 

diff --git a/hmda/src/main/resources/application.conf b/hmda/src/main/resources/application.conf
@@ -1,4 +1,10 @@
+# Apache Cassandra with PlainTextAuthProvider
 include "persistence.conf"
+# AWS Keyspace with PlainTextAuthProvider
+# include "persistence-keyspace.conf"
+# AWS Keyspace with SigV4AuthProvider
+# include "persistence-keyspace-sigv4.conf"
+
 include "hmda.conf"
 include "cors.conf"
 include "edits.conf"

diff --git a/irs-publisher/src/main/resources/application.conf b/irs-publisher/src/main/resources/application.conf
@@ -1,4 +1,9 @@
+# Apache Cassandra with PlainTextAuthProvider
 include "persistence.conf"
+# AWS Keyspace with PlainTextAuthProvider
+# include "persistence-keyspace.conf"
+# AWS Keyspace with SigV4AuthProvider
+# include "persistence-keyspace-sigv4.conf"
 
 akka {
   loglevel = info