Test with AWS Keyspaces

patriknw commented 3 years ago

We know some differences:

Keyspaces has different semantics and performance characteristics for unlogged batches
- they aren't a performance optimization
- they aren't Atomic/Isolated
imposes a 1k operation / second on any given Cassandra partition. Hitting this limit of 1k operations/s translate to a timeout exception given to the client. Whereby an operation is either:
- 1kb of write (so max 1mb/s)
- A write to a column

patriknw commented 3 years ago

Notes of my testing of Akka Persistence Cassandra with Keyspaces.

Functional Differences: Amazon Keyspaces (for Apache Cassandra) versus Apache Cassandra: https://docs.aws.amazon.com/keyspaces/latest/devguide/functional-differences.html

Supported consistency levels: https://docs.aws.amazon.com/keyspaces/latest/devguide/consistency.html

Setup

Console: https://eu-central-1.console.aws.amazon.com/keyspaces/home?region=eu-central-1#service

Create truststore according to: https://docs.aws.amazon.com/keyspaces/latest/devguide/using_java_driver.html#using_java_driver.BeforeYouBegin

This truststore is used via the parameters: -Djavax.net.ssl.trustStore=path_to_file/cassandra_truststore.jks -Djavax.net.ssl.trustStorePassword=amazon The password is what you used when creating the truststore.

Create IAM user according to: https://docs.aws.amazon.com/keyspaces/latest/devguide/programmatic.credentials.html#programmatic.credentials.ssc

There are instructions to create credentials for keyspaces, but I never needed those later. I used the Access key id for the IAM user as environment variables. export AWS_ACCESS_KEY=AKIA3WD6YIW5BUIBWFHC export AWS_SECRET_ACCESS_KEY=

These environment variables are not fully documented but the error message is pretty clear if they are not defined.

Add the authentication plugin: https://docs.aws.amazon.com/keyspaces/latest/devguide/using_java_driver.html#java_tutorial.SigV4 "software.aws.mcs" % "aws-sigv4-auth-cassandra-java-driver-plugin" % "4.0.2",

Config

Configure the authentication plugin in application.conf (CassandraLifecycle.config):

    datastax-java-driver { 
      basic.contact-points = [ "cassandra.eu-central-1.amazonaws.com:9142"]
      basic.request.consistency = LOCAL_QUORUM
      basic.load-balancing-policy {
        class = DefaultLoadBalancingPolicy
        local-datacenter = eu-central-1
      }
      advanced {
        auth-provider = {
          class = software.aws.mcs.auth.SigV4AuthProvider
          aws-region = eu-central-1
        }
        ssl-engine-factory {
          class = DefaultSslEngineFactory
        }
      }
    }

Note that the contact points must also be configured, which isn’t fully documented. You find the endpoints for different regions at https://docs.aws.amazon.com/keyspaces/latest/devguide/programmatic.endpoints.html

patriknw commented 3 years ago

Test results

It was difficult to run tests because keyspace and table creation are asynchronous and not finished when the returned Future is completed. In the end I abandoned the automatic creation and created/dropped the tables with cql script via cqlsh.

Successful tests:

akka.persistence.cassandra.journal.BufferSpec
akka.persistence.cassandra.journal.CassandraJournalMetaSpec
akka.persistence.cassandra.journal.PubSubThrottlerSpec
akka.persistence.cassandra.journal.TagWriterSpec
akka.persistence.cassandra.journal.CassandraEventUpdateSpec
akka.persistence.cassandra.journal.CassandraSerializationSpec
akka.persistence.cassandra.journal.TagWritersSpec
akka.persistence.cassandra.journal.CassandraIntegrationSpec
akka.persistence.cassandra.journal.CassandraJournalSpec
akka.persistence.cassandra.journal.TimeBucketSpec
akka.persistence.cassandra.journal.CassandraJournalDeletionSpec
akka.persistence.cassandra.journal.TagScanningSpec
akka.persistence.cassandra.EventsByTagRestartSpec
akka.persistence.cassandra.RetriesSpec
akka.persistence.cassandra.healthcheck.CassandraHealthCheckCustomFailingQuerySpec
akka.persistence.cassandra.CassandraCorruptJournalSpec
akka.persistence.cassandra.CassandraQueryJournalSettingsSpec
akka.persistence.cassandra.EventsByTagCrashSpec
akka.persistence.cassandra.reconciler.TagQuerySpec
akka.persistence.cassandra.reconciler.DeleteTagViewForPersistenceIdSpec
akka.persistence.cassandra.sharding.ClusterShardingQuickTerminationSpec
akka.persistence.cassandra.query.EventsByPersistenceIdFastForwardSpec
akka.persistence.cassandra.query.EventsByPersistenceIdWithControlSpec
akka.persistence.cassandra.query.EventsByTagLongRefreshIntervalSpec
akka.persistence.cassandra.query.EventsByTagSpec
akka.persistence.cassandra.query.EventsByTagStageSpec
akka.persistence.cassandra.query.EventsByTagStrictBySeqNoManyInCurrentTimeBucketSpec
akka.persistence.cassandra.query.javadsl.CassandraReadJournalSpec
akka.persistence.cassandra.query.scaladsl.CassandraReadJournalSpec
akka.persistence.cassandra.query.CassandraQueryJournalOverrideSpec 
akka.persistence.cassandra.query.EventsByPersistenceIdMultiPartitionGapSpec
akka.persistence.cassandra.query.EventsByTagDisabledSpec
akka.persistence.cassandra.query.EventsByTagSpecBackTracking
akka.persistence.cassandra.query.EventsByTagStrictBySeqMemoryIssueSpec
akka.persistence.cassandra.query.EventsByTagZeroEventualConsistencyDelaySpec (282 s)
akka.persistence.cassandra.query.EventsByPersistenceIdSpec
akka.persistence.cassandra.query.EventAdaptersReadSpec
akka.persistence.cassandra.query.EventsByTagSpecBackTrackingLongRefreshInterval
akka.persistence.cassandra.query.EventsByTagStrictBySeqNoEarlyFirstOffsetSpec (85 s)

Failing test (in order of importance):

[x] akka.persistence.cassandra.query.EventsByTagFindDelayedEventsSpec
- find delayed events when many other events
- got unexpected message
[x] akka.persistence.cassandra.healthcheck.CassandraHealthCheckDefaultQuerySpec
- changed: health-check-cql = "SELECT now() FROM system.local" => health-check-cql = "SELECT * FROM system.local"
- but there are also other failures
[x] akka.persistence.cassandra.cleanup.CleanupSpec
[x] akka.persistence.cassandra.snapshot.CassandraSnapshotStoreSpec
- snapshots that cannot be de-serialized
- Attempted to set an element with an invalid value
- but that is written from the test itself
[ ] akka.persistence.cassandra.query.EventsByTagPersistenceIdCleanupSpec
[ ] akka.persistence.cassandra.compaction.CassandraCompactionStrategySpec
- compaction is not releveant for Keyspaces
[ ] akka.persistence.cassandra.reconciler.RebuildAllPersisetceIdsSpec
- DISTINCT is not yet supported
- only used by currentPersistenceIdsFromMessages via Reconciliation.rebuildAllPersistenceIds
[ ] akka.persistence.cassandra.healthcheck.CassandraHealthCheckCustomQueryNonEmptyResultSpec
- system keyspace is not user-modifiable
[ ] akka.persistence.cassandra.ReconnectSpec
- requires localhost connect
[ ] akka.persistence.cassandra.journal.MultiPluginSpec
[ ] akka.persistence.cassandra.EventsByTagMigrationSpec

Failing becuse truncate not supported

[x] akka.persistence.cassandra.healthcheck.CassandraHealthCheckCustomQueryEmptyResultSpec
[ ] akka.persistence.cassandra.reconciler.BuildTagViewForPersisetceIdSpec
[ ] akka.persistence.cassandra.reconciler.TruncateAllSpec
[x] akka.persistence.cassandra.query.TagViewSequenceNumberScannerSpec
[ ] akka.persistence.cassandra.query.AllPersistenceIdsSpec

Performance related tests that I didn't try much (they probably fail due to throttling):

akka.persistence.cassandra.journal.CassandraLoadTypedSpec
akka.persistence.cassandra.journal.CassandraJournalPerfSpec
akka.persistence.cassandra.journal.RecoveryLoadSpec
akka.persistence.cassandra.journal.ManyActorsLoadSpec
akka.persistence.cassandra.journal.StartupLoadSpec
akka.persistence.cassandra.journal.CassandraLoadSpec
akka.persistence.cassandra.EventsByTagStressSpec 
akka.persistence.cassandra.CassandraEventsByTagLoadSpec