Bump aws version and add query

Author: pbernet

build.sbt

@@ -18,7 +18,7 @@ val testContainersVersion = "1.19.7"
 val keycloakVersion = "21.1.2" // stay with 21.x because of Java 11 compatibility
 val sttpVersion = "3.9.0"
 val influxdbVersion = "6.10.0"
-val awsClientVersion = "2.25.21"
+val awsClientVersion = "2.25.32"
 
 libraryDependencies ++= Seq(
   "org.scala-lang.modules" %% "scala-parallel-collections" % "1.0.4",

src/main/scala/alpakka/dynamodb/DynamoDBEcho.scala

@@ -40,7 +40,7 @@ class DynamoDBEcho(urlWithMappedPort: URI, accessKey: String, secretKey: String,
     for {
       _ <- createTable()
       _ <- writeItems(noOfItems)
-      result <- readItems()
+      result <- readItems(noOfItems)
     } yield result
   }
 
@@ -53,7 +53,7 @@ class DynamoDBEcho(urlWithMappedPort: URI, accessKey: String, secretKey: String,
       .via(DynamoDb.flow(parallelism = 1))
 
     source
-      .map(descTableResponse => logger.info(s"Successfully created table: ${descTableResponse.table.tableName}"))
+      .wireTap(descTableResponse => logger.info(s"Successfully created table: ${descTableResponse.table.tableName}"))
       .runWith(Sink.ignore)
   }
 
@@ -95,7 +95,7 @@ class DynamoDBEcho(urlWithMappedPort: URI, accessKey: String, secretKey: String,
         val request = PutItemRequest.builder().tableName(testTableName).item(Map(
           "Id" -> AttributeValue.builder().s(item.toString).build(),
           "att1" -> AttributeValue.builder().s(s"att1-$item").build(),
-          "att2" -> AttributeValue.builder().s(s"att2-$item").build()
+          "att2" -> AttributeValue.builder().n(s"$item").build()
         ).asJava).build()
         (request, RequestContext(testTableName, requestId))
       })
@@ -114,23 +114,34 @@ class DynamoDBEcho(urlWithMappedPort: URI, accessKey: String, secretKey: String,
       .runWith(Sink.ignore)
   }
 
-  private def readItems() = {
-    logger.info(s"About to read items...")
+  private def readItems(noOfItems: Int) = {
+    logger.info(s"About to read 2nd half of all items...")
+
+    val filterExpression = "#att2 > :val"
+    val expressionAttrNames = new java.util.HashMap[String, String]()
+    expressionAttrNames.put("#att2", "att2")
+
+    val expressionAttrValues = new java.util.HashMap[String, AttributeValue]()
+    expressionAttrValues.put(":val", AttributeValue.builder().n((noOfItems / 2).toString).build())
+
+    val scanRequest = ScanRequest.builder()
+      .tableName(testTableName) // This hangs when no table is available
+      .filterExpression(filterExpression)
+      .expressionAttributeNames(expressionAttrNames)
+      .expressionAttributeValues(expressionAttrValues)
+      .build()
 
-    // This hangs when no table is available
-    val scanRequest = ScanRequest.builder().tableName(testTableName).build()
     val scanPageInFlow: Source[ScanResponse, NotUsed] =
       Source
         .single(scanRequest)
         .via(DynamoDb.flowPaginated())
 
-    scanPageInFlow.map { (response: ScanResponse) => {
-      val count = response.scannedCount()
-      logger.info(s"Successfully read $count items")
-      response.items().forEach(item => logger.info(s"Item: $item"))
-      response.items().size()
-    }
-
+    scanPageInFlow.map { scanResponse =>
+      val count = scanResponse.scannedCount()
+      val resultCount = scanResponse.items().size()
+      logger.info(s"Successfully read $resultCount/$count items")
+      scanResponse.items().forEach(item => logger.info(s"Item: $item"))
+      resultCount
     }.runWith(Sink.head)
   }
 

src/test/scala/alpakka/dynamodb/DynamoDBEchoIT.java

@@ -20,6 +20,12 @@
 
 /**
  * Setup/run {@link alpakka.dynamodb.DynamoDBEcho} on localStack container
+ *
+ * Running this example against AWS:
+ * Looks as if there is a way to delete a DB instance via the SDK:
+ * https://docs.aws.amazon.com/code-library/latest/ug/rds_example_rds_DeleteDBInstance_section.html
+ * However, getting the `dbInstanceIdentifier` via SDK is not straightforward
+ * Therefore, we only run against localStack for now in order to avoid dangling resources
  * <p>
  * Doc:
  * https://pekko.apache.org/docs/pekko-connectors/current/dynamodb.html#aws-dynamodb
@@ -45,10 +51,10 @@ public static void beforeAll() {
     @Test
     public void testLocal() throws ExecutionException, InterruptedException {
         DynamoDBEcho dynamoDBEcho = new DynamoDBEcho(localStack.getEndpointOverride(DYNAMODB), localStack.getAccessKey(), localStack.getSecretKey(), localStack.getRegion());
-        int noOfItems = 10;
+        int noOfItemsEven = 10;
 
-        CompletionStage<Object> result = FutureConverters.asJava(dynamoDBEcho.run(noOfItems));
-        assertThat(result.toCompletableFuture().get()).isEqualTo(noOfItems);
+        CompletionStage<Object> result = FutureConverters.asJava(dynamoDBEcho.run(noOfItemsEven));
+        assertThat(result.toCompletableFuture().get()).isEqualTo(noOfItemsEven / 2);
     }
 }
 

src/test/scala/alpakka/kinesis/KinesisEchoIT.java

@@ -28,7 +28,7 @@
 
 /**
  * Setup/run {@link alpakka.kinesis.KinesisEcho} on localStack container
- * Additionally use the classic sync AWS KinesisClient to create/delete streams
+ * Use the classic sync AWS KinesisClient to create/delete streams
  * <p>
  * Doc:
  * https://testcontainers.com/modules/localstack