Mouwrice commented on code in PR #349:
URL:
https://github.com/apache/pekko-persistence-dynamodb/pull/349#discussion_r3712192849
##########
README.md:
##########
@@ -6,10 +6,17 @@ A replicated Pekko Persistence journal backed by
- This plugin implements both a journal as well as a snapshot store,
- This includes a Pekko Persistence Query plugin. However, this requires an
additional GSI for efficient usage.
-Supported versions:
-- Scala: `2.13.x`, `3.3.0+`
+Apache Pekko Persistence DynamoDB 1.x Supported versions:
+- Scala: `2.12.x`, `2.13.x`, `3.3.0+`
- Pekko: `1.0.x+`
+- Java: `8+`
+- Amazon SDK for Java v1
+
+Apache Pekko Persistence DynamoDB 2.x Supported versions:
+- Scala: `2.13.x`, `3.3.0+`
+- Pekko: `2.0.x+`
- Java: `17+`
+- Amazon SDK for Java v2
Review Comment:
for version 1.x this adds scala 2.12 (not true)
also lowers the java version from 17 to 8
both 1.x and 2.x also support scala 3.8.x
##########
README.md:
##########
Review Comment:
When this PR is close to being merged, probably best to add some migration
guidance here or in the release notes.
##########
build.sbt:
##########
@@ -10,7 +10,7 @@
import com.github.pjfanning.pekkobuild._
import
net.bzzt.reproduciblebuilds.ReproducibleBuildsPlugin.reproducibleBuildsCheckResolver
-val amzVersion = "1.12.797"
+val amzV2Version = "2.47.5"
Review Comment:
Newer versions already exists, should this be updated frequently?
##########
src/main/scala/org/apache/pekko/persistence/dynamodb/DynamoDBConfig.scala:
##########
@@ -37,50 +33,4 @@ trait DynamoDBConfig {
}
-class DynamoDBClientConfig(c: Config) extends ClientConfig {
- private val cc = c.getConfig("aws-client-config")
- private def get[T](path: String, extract: (Config, String) => T, set: T =>
Unit): Unit =
- if (cc.getString(path) == "default") ()
- else {
- val value = extract(cc, path)
- set(value)
- foundSettings ::= s"$path:$value"
- }
-
- private var foundSettings = List.empty[String]
- override lazy val toString: String = foundSettings.reverse.mkString("{",
",", "}")
- val config = new ClientConfiguration
-
- get("client-execution-timeout", _.getInt(_),
config.setClientExecutionTimeout)
- get("connection-max-idle-millis", _.getLong(_),
config.setConnectionMaxIdleMillis)
- get("connection-timeout", _.getInt(_), config.setConnectionTimeout)
- get("connection-ttl", _.getLong(_), config.setConnectionTTL)
- get("local-address", (c, p) => InetAddress.getByName(c.getString(p)),
config.setLocalAddress)
- get("max-connections", _.getInt(_), config.setMaxConnections)
- get("max-error-retry", _.getInt(_), config.setMaxErrorRetry)
- get("preemptive-basic-proxy-auth", _.getBoolean(_),
config.withPreemptiveBasicProxyAuth)
- get("protocol", (c, p) => if (c.getString(p) == "HTTP") Protocol.HTTP else
Protocol.HTTPS, config.setProtocol)
- get("proxy-domain", _.getString(_), config.setProxyDomain)
- get("proxy-host", _.getString(_), config.setProxyHost)
- get("proxy-password", _.getString(_), config.setProxyPassword)
- get("proxy-port", _.getInt(_), config.setProxyPort)
- get("proxy-username", _.getString(_), config.setProxyUsername)
- get("proxy-workstation", _.getString(_), config.setProxyWorkstation)
- get("request-timeout", _.getInt(_), config.setRequestTimeout)
- get("response-metadata-cache-size", _.getInt(_),
config.setResponseMetadataCacheSize)
- get("signer-override", _.getString(_), config.setSignerOverride)
- get[(Int, Int)](
- "socket-buffer-size-hints",
- (c, p) => {
- val tuple = c.getIntList(p)
- require(tuple.size == 2, "socket-buffer-size-hints must be a list of two
integers")
- (tuple.get(0), tuple.get(1))
- },
- pair => config.setSocketBufferSizeHints(pair._1, pair._2))
- get("socket-timeout", _.getInt(_), config.setSocketTimeout)
- get("use-expect-continue", _.getBoolean(_), config.setUseExpectContinue)
- get("use-gzip", _.getBoolean(_), config.setUseExpectContinue)
- get("use-reaper", _.getBoolean(_), config.setUseReaper)
- get("use-tcp-keepalive", _.getBoolean(_), config.setUseTcpKeepAlive)
- get("user-agent", _.getString(_), config.setUserAgentPrefix)
-}
+class DynamoDBClientConfig(c: Config) extends ClientConfig
Review Comment:
The client can no longer be configure via the config file, so not much use
to keep this around any longer (`ClientConfig`, `DynamoDBClientConfig`)
##########
src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBHelper.scala:
##########
@@ -135,103 +123,59 @@ trait DynamoDBHelper {
f
}
- trait Describe[T] {
- def desc(t: T): String
- protected def formatKey(i: Item): String = {
- val key = i.get(Key) match {
- case null => "<none>"
- case x => x.getS
- }
- val sort = i.get(Sort) match {
- case null => "<none>"
- case x => x.getN
- }
- s"[$Key=$key,$Sort=$sort]"
+ protected def formatKey(i: Item): String = {
+ val key = i.get(Key) match {
+ case null => "<none>"
+ case x => x.s
}
- }
-
- object Describe {
- implicit object GenericDescribe extends Describe[AmazonWebServiceRequest] {
- def desc(aws: AmazonWebServiceRequest): String =
aws.getClass.getSimpleName
+ val sort = i.get(Sort) match {
+ case null => "<none>"
+ case x => x.n
}
+ s"[$Key=$key,$Sort=$sort]"
}
- implicit object DescribeDescribe extends Describe[DescribeTableRequest] {
- def desc(aws: DescribeTableRequest): String =
s"DescribeTableRequest(${aws.getTableName})"
- }
+ def listTables(aws: ListTablesRequest): Future[ListTablesResponse] =
+ send(s"ListTablesRequest", dynamoDB.listTables(aws).asScala)
- implicit object QueryDescribe extends Describe[QueryRequest] {
- def desc(aws: QueryRequest): String =
s"QueryRequest(${aws.getTableName},${aws.getExpressionAttributeValues})"
- }
+ def describeTable(aws: DescribeTableRequest): Future[DescribeTableResponse] =
+ send(s"DescribeTableRequest(${aws.tableName})",
dynamoDB.describeTable(aws).asScala)
- implicit object PutItemDescribe extends Describe[PutItemRequest] {
- def desc(aws: PutItemRequest): String =
s"PutItemRequest(${aws.getTableName},${formatKey(aws.getItem)})"
- }
+ def createTable(aws: CreateTableRequest): Future[CreateTableResponse] =
+ send(s"CreateTableRequest(${aws.tableName})",
dynamoDB.createTable(aws).asScala)
- implicit object DeleteDescribe extends Describe[DeleteItemRequest] {
- def desc(aws: DeleteItemRequest): String =
s"DeleteItemRequest(${aws.getTableName},${formatKey(aws.getKey)})"
- }
+ def updateTable(aws: UpdateTableRequest): Future[UpdateTableResponse] =
+ send(s"UpdateTableRequest(${aws.tableName})",
dynamoDB.updateTable(aws).asScala)
- implicit object BatchGetItemDescribe extends Describe[BatchGetItemRequest] {
- def desc(aws: BatchGetItemRequest): String = {
- val entry = aws.getRequestItems.entrySet.iterator.next()
- val table = entry.getKey
- val keys = entry.getValue.getKeys.asScala.map(formatKey)
- s"BatchGetItemRequest($table, ${keys.mkString("(", ",", ")")})"
- }
- }
-
- implicit object BatchWriteItemDescribe extends
Describe[BatchWriteItemRequest] {
- def desc(aws: BatchWriteItemRequest): String = {
- val entry = aws.getRequestItems.entrySet.iterator.next()
- val table = entry.getKey
- val keys = entry.getValue.asScala.map { write =>
- write.getDeleteRequest match {
- case null => "put" + formatKey(write.getPutRequest.getItem)
- case del => "del" + formatKey(del.getKey)
- }
- }
- s"BatchWriteItemRequest($table, ${keys.mkString("(", ",", ")")})"
- }
- }
+ def deleteTable(aws: DeleteTableRequest): Future[DeleteTableResponse] =
+ send(s"DeleteTableRequest(${aws.tableName})",
dynamoDB.deleteTable(aws).asScala)
- def listTables(aws: ListTablesRequest): Future[ListTablesResult] =
- send[ListTablesRequest, ListTablesResult](aws,
dynamoDB.listTablesAsync(aws, _))
+ def query(aws: QueryRequest): Future[QueryResponse] =
+ send(s"QueryRequest(${aws.tableName},${aws.expressionAttributeValues})",
dynamoDB.query(aws).asScala)
- def describeTable(aws: DescribeTableRequest): Future[DescribeTableResult] =
- send[DescribeTableRequest, DescribeTableResult](aws,
dynamoDB.describeTableAsync(aws, _))
+ def scan(aws: ScanRequest): Future[ScanResponse] =
+ send(s"ScanRequest(${aws.tableName})", dynamoDB.scan(aws).asScala)
- def createTable(aws: CreateTableRequest): Future[CreateTableResult] =
- send[CreateTableRequest, CreateTableResult](aws,
dynamoDB.createTableAsync(aws, _))
+ def putItem(aws: PutItemRequest): Future[PutItemResponse] =
+ send(s"PutItemRequest(${aws.tableName},${formatKey(aws.item)})",
dynamoDB.putItem(aws).asScala)
- def updateTable(aws: UpdateTableRequest): Future[UpdateTableResult] =
- send[UpdateTableRequest, UpdateTableResult](aws,
dynamoDB.updateTableAsync(aws, _))
+ def getItem(aws: GetItemRequest): Future[GetItemResponse] =
+ send(s"GetItemRequest(${aws.tableName})", dynamoDB.getItem(aws).asScala)
- def deleteTable(aws: DeleteTableRequest): Future[DeleteTableResult] =
- send[DeleteTableRequest, DeleteTableResult](aws,
dynamoDB.deleteTableAsync(aws, _))
+ def updateItem(aws: UpdateItemRequest): Future[UpdateItemResponse] =
+ send(s"UpdateItemRequest(${aws.tableName})",
dynamoDB.updateItem(aws).asScala)
- def query(aws: QueryRequest): Future[QueryResult] =
- send[QueryRequest, QueryResult](aws, dynamoDB.queryAsync(aws, _))
+ def deleteItem(aws: DeleteItemRequest): Future[DeleteItemResponse] =
+ send(s"DeleteItemRequest(${aws.tableName},${formatKey(aws.key)})",
dynamoDB.deleteItem(aws).asScala)
- def scan(aws: ScanRequest): Future[ScanResult] =
- send[ScanRequest, ScanResult](aws, dynamoDB.scanAsync(aws, _))
+ def batchWriteItem(aws: BatchWriteItemRequest):
Future[BatchWriteItemResponse] =
+ send(s"BatchWriteItemRequest(${aws.requestItems.keySet.iterator.next})",
dynamoDB.batchWriteItem(aws).asScala)
Review Comment:
I believe the batch write description does not match the original
implementation in BatchWriteItemDescribe
##########
src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBHelper.scala:
##########
@@ -67,46 +65,36 @@ trait DynamoDBHelper {
implicit val ec: ExecutionContext
val scheduler: Scheduler
- val dynamoDB: AmazonDynamoDBAsync
+ val dynamoDB: DynamoDbAsyncClient
val log: LoggingAdapter
val settings: DynamoDBConfig
import settings._
- def shutdown(): Unit = dynamoDB.shutdown()
+ def shutdown(): Unit = dynamoDB.close()
- private var reporter: ActorRef = null
+ private var reporter: ActorRef = _
def setReporter(ref: ActorRef): Unit = reporter = ref
- private def send[In <: AmazonWebServiceRequest, Out](aws: In, func:
AsyncHandler[In, Out] => juc.Future[Out])(implicit
- d: Describe[? >: In]): Future[Out] = {
-
- def name = d.desc(aws)
+ private def send[Out](name: => String, call: => Future[Out]): Future[Out] = {
def sendSingle(): Future[Out] = {
- val p = Promise[Out]()
-
- val handler = new AsyncHandler[In, Out] {
- override def onError(ex: Exception) =
- ex match {
- case DynamoRetriableException(_) =>
- p.tryFailure(ex)
- case _ =>
- val n = name
- log.error(ex, "failure while executing {}", n)
- p.tryFailure(new DynamoDBJournalFailure("failure while executing
" + n, ex))
+ call.recoverWith {
+ case ex: DynamoDbException if
DynamoRetriableException.unapply(ex).isEmpty =>
Review Comment:
```suggestion
case DynamoRetriableException(ex) =>
```
##########
src/test/scala/org/apache/pekko/persistence/dynamodb/journal/FailureReportingSpec.scala:
##########
@@ -204,62 +204,45 @@ pekko.loggers =
["org.apache.pekko.testkit.TestEventListener"]
}
"have sensible error messages" when {
- val evaluatedDynamo = dynamo
- import evaluatedDynamo._
- def desc[T](aws: T)(implicit d: Describe[? >: T]): String = d.desc(aws)
-
val keyItem = Map(Key -> S("TheKey"), Sort -> N("42")).asJava
val key2Item = Map(Key -> S("The2Key"), Sort -> N("43")).asJava
"reporting table problems" in {
- val aws = new DescribeTableRequest().withTableName("TheTable")
- desc(aws) should include("DescribeTable")
- desc(aws) should include("TheTable")
+ val aws = DescribeTableRequest.builder().tableName("TheTable").build()
+ dynamo.describeTable(aws).failed.futureValue.getMessage should
+ (include("DescribeTable").or(include("TheTable")))
}
"reporting putItem problems" in {
- val aws = new
PutItemRequest().withTableName("TheTable").withItem(keyItem)
- desc(aws) should include("PutItem")
- desc(aws) should include("TheTable")
- desc(aws) should include("TheKey")
- desc(aws) should include("42")
+ val aws =
PutItemRequest.builder().tableName("TheTable").item(keyItem).build()
+ dynamo.putItem(aws).failed.futureValue.getMessage should
(include("PutItem").or(include("TheTable")))
Review Comment:
The tests now assert inclusion of at least one of these, instead of checking
all of them. Use `.and` instead of `.or`
##########
src/main/resources/reference.conf:
##########
@@ -81,37 +88,11 @@ dynamodb-journal {
max-item-size = 400000
}
- # AWS client configuration settings, see
- #
https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/?com/amazonaws/ClientConfiguration.html
- #
- # (setting any of these to the string "default" means that the corresponding
- # setter method on the ClientConfiguration will not be invoked)
+ # AWS client configuration settings.
+ # The AWS SDK v2 DynamoDbAsyncClient is configured via its builder in code.
+ # For advanced HTTP client settings, provide a custom DynamoDbAsyncClient.
+ # See
https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/http-configuration.html
aws-client-config {
Review Comment:
Wondering if this should be kept in the config or in some migration notes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
