AWS SDK 2 for Java and storing a Json in DynamoDB

 AWS DynamoDB is described as a NoSQL key-value and a document database. In my work I mostly use the key-value behavior of the database but rarely use the document database features, however  the document database part is growing on me and this post highlights some ways of using the document database feature of DynamoDB along with introducing a small utility library built on top of AWS SDK 2.X for Java that simplifies using document database features of AWS DynamoDB

The treatment of the document database features will be very high level in this post, I will plan a follow up which goes into more details later

DynamoDB as a document database

So what does it mean for AWS DynamoDB to be treated as a document database. Consider a json representation of an entity, say something representing a Hotel:

{
    "id": "1",
    "name": "test",
    "address": "test address",
    "state": "OR",
    "properties": {
        "amenities":{
            "rooms": 100,
            "gym": 2,
            "swimmingPool": true                    
        }                    
    },
    "zip": "zip"
}

This json has some top level attributes like "id", a name, an address etc. But it also has a free form "properties" holding some additional "nested" attributes of this hotel. 

A document database can store this document representing the hotel in its entirety OR can treat individual fields say the "properties" field of the hotel as a document. 

A naive way to do this will be to simply serialize the entire content into a json string and store it in, say for eg, for the properties field transform into a string representation of the json and store in the database, this works, but there are a few issues with it. 

1. None of the attributes of the field like properties can be queried for, say if I wanted to know whether the hotel has a swimming pool, there is no way just to get this information of of the stored content. 
2. The attributes cannot be filtered on - so say if wanted hotels with atleast 2 gyms, this is not something that can be filtered down to. 

A document database would allow for the the entire document to be saved, individual attributes, both top level and nested ones, to be queried/filtered on. 

So for eg, in the example of "hotel" document the top level attributes are "id", "name", "address", "state", "zip" and the nested attributes are "properties.amenities.rooms", "properties.amenities.gym", "properties.amenities.swimmingPool" and so on.

AWS SDK 2 for DynamoDB and Document database support

If you are writing a Java based application to interact with a AWS DynamoDB database, then you would have likely used the new AWS SDK 2 library to make the API calls. However one issue with the library is that it natively does not support a json based document model. Let me go into a little more detail here.  

AWS DynamoDB version field using AWS SDK for Java 2

It is useful to have a version attribute on any entity saved to an AWS DynamoDB database which is simply a numeric indication of the number of times the entity has been modified. When the entity is first created it can be set to 1 and then incremented on every update. 

The benefit is immediate - an indicator of the number of times an entity has been modified which can be used for auditing the entity. Also, an additional use is for optimistic locking where an entity is allowed to be updated only if the holder updating it has the right version of the entity.

This post will go into details of how to introduce such a field with the DynamoDB related libraries of AWS SDK 2

Model

Consider a model called Hotel which is being persisted into a dynamo database. In Kotlin, it can be represented using the following data class:

data class Hotel(
    val id: String = UUID.randomUUID().toString(),
    val name: String,
    val address: String? = null,
    val state: String? = null,
    val zip: String? = null,
    val version: Long = 1L
)

A version field has been introduced in this model with an initial value of 1. The aim will be to save this field as-is and then let dynamo atomically manage the increment of this field at the point of saving this entity.

As the fields in this model gets changed, I would like the version to be updated along these lines:

 

Local version of DynamoDB

It is useful to have DynamoDB running on the local machine, this way not having to create real DynamoDB tables in AWS.

There are multiple ways of doing this. One is to use a docker version of DynamoDB Local, which can be started up the following way to listen on port 4569:

docker run -p 4569:8000 amazon/dynamodb-local:1.13

My personal preference is to use localstack and the instructions at the site have different ways to start it up. I normally use docker-compose to bring it up. One of the reasons to use localstack over DynamoDB Local is that localstack provides a comprehensive set of AWS services for local testing and not just DynamoDB.

Quick Demo

I have the entire code available in my github repo here - https://github.com/bijukunjummen/boot-with-dynamodb

Once the application is brought up using the local version of dynamoDB, an entity can be created using the following httpie request:

http :9080/hotels id=4 name=name address=address zip=zip state=OR

With a response, where the version field is set to 1:

{
    "address": "address",
    "id": "4",
    "name": "name",
    "state": "OR",
    "version": 1,
    "zip": "zip"
}

Then if the name is updated for the entity:

http PUT :9080/hotels/4 name=name1 address=address zip=zip state=OR version=1

the version field gets updated to 2 and so on:

{
    "address": "address",
    "id": "4",
    "name": "name1",
    "state": "OR",
    "version": 2,
    "zip": "zip"
}

Also note that if during an update a wrong version number is provided, the call would fail as there is an optimistic locking in place using this version field.

Implementing the version field

Implementing the version field depends on the powerful UpdateItem API provided by DynamoDB. One of the features of UpdateItem API is that it takes in a "UpdateExpression" which is a dsl which shows how different Dynamo attributes should be updated.

The raw request to AWS DynamoDB looks like this:

 

{
  "TableName": "hotels",
  "Key": {
    "id": {
      "S": "1"
    }
  },
  "UpdateExpression": "\nSET #name=:name,\n #state=:state,\naddress=:address,\nzip=:zip\nADD version :inc\n ",
  "ExpressionAttributeNames": {
    "#state": "state",
    "#name": "name"
  },
  "ExpressionAttributeValues": {
    ":name": {
      "S": "testhotel"
    },
    ":address": {
      "S": "testaddress"
    },
    ":state": {
      "S": "OR"
    },
    ":zip": {
      "S": "zip"
    },
    ":inc": {
      "N": "1"
    }
  }
}               

From the articles perspective, specifically focus on "ADD version :inc", which is an expression that tells AWS DynamoDB to increment the value of version by ":inc" value, which is provided separately using "ExpressionAttributeValues" with "1". Dealing with raw API in its json form is daunting, that is where the Software Development Kit(SDK) that AWS provides comes in, AWS SDK for Java 2 is a rewrite of AWS SDK's with a focus on using the latest Java features and Non-Blocking IO over the wire. Using AWS SDK for Java 2, an "UpdateItem" looks like this(using Kotlin code):

val updateItemRequest = UpdateItemRequest.builder()
    .tableName(TABLE_NAME)
    .key(
        mapOf(
            ID to AttributeValue.builder().s(hotel.id).build()
        )
    )
    .updateExpression(
    """
        SET #name=:name,
        #state=:state,
        address=:address,
        zip=:zip 
        ADD version :inc
    """
    )
    .conditionExpression("version = :version")
    .expressionAttributeValues(
        mapOf(
            ":${NAME}" to AttributeValue.builder().s(hotel.name).build(),
            ":${ZIP}" to AttributeValue.builder().s(hotel.zip).build(),
            ":${STATE}" to AttributeValue.builder().s(hotel.state).build(),
            ":${ADDRESS}" to AttributeValue.builder().s(hotel.address).build(),
            ":${VERSION}" to AttributeValue.builder().n(hotel.version.toString()).build(),
            ":inc" to AttributeValue.builder().n("1").build()
        )
    )
    .expressionAttributeNames(
        mapOf(
            "#name" to "name",
            "#state" to "state"
        )
    )
    .build()

val updateItem: CompletableFuture<UpdateItemResponse> = dynamoClient.updateItem(updateItemRequest)

return Mono.fromCompletionStage(updateItem)
    .flatMap {
        getHotel(hotel.id)
    }

The highlighted line has the "Update Expression" with all the existing fields set to a new value and the version attribute incremented by 1. Another thing to note about this call is the "conditionExpression", which is essentially a way to tell DynamoDB to update the attributes if a condition matches up, in this specific instance if the existing value of version matches up. This provides a neat way to support optimistic locking on the record.

Conclusion

A lot of details here - the easiest way to get a feel for it is by trying out the code which is available in my github repository here - https://github.com/bijukunjummen/boot-with-dynamodb. The readme has good details on how to run it in a local environment. 

AWS DynamoDB provides a neat way to manage a version field on entities, ensuring that they are atomically updated and provides a way for them to used for optimistic locking

Processing SQS Messages using Spring Boot and Project Reactor

I recently worked on a project where I had to efficiently process a large number of messages streaming in through an AWS SQS Queue. In this post (and potentially one more), I will go over the approach that I took to process the messages using the excellent Project Reactor



The following is the kind of set-up that I am aiming for:

Setting up a local AWS Environment

Before I jump into the code, let me get some preliminaries out of the way. First, how do you get a local version of SNS and SQS. One of the easiest ways is to use localstack. I use a docker-compose version of it described here

The second utility that I will be using is the AWS CLI. This website has details on how to install it locally.

Once both of these utilities are in place, a quick test should validate the setup:

# Create a queue
aws --endpoint http://localhost:4576 sqs create-queue --queue-name test-queue

# Send a sample message
aws --endpoint http://localhost:4576 sqs send-message --queue-url http://localhost:4576/queue/test-queue --message-body "Hello world"

# Receive the message
aws --endpoint http://localhost:4576 sqs receive-message --queue-url http://localhost:4576/queue/test-queue

Basics of Project Reactor

Project Reactor implements the Reactive Streams specification and provides a way of handling streams of data across asynchronous boundaries that respects backpressure. A lot of words here but in essence think of it this way:
1. SQS Produces data
2. The application is going to consume and process it as a stream of data
3. The application should consume data at a pace that is sustainable - too much data should not be pumped in. This is formally referred to as "Backpressure"

AWS SDK 2

The library that I will be using to consume AWS SQS data is the AWS SDK 2. The library uses non-blocking IO under the covers.

The library offers both a sync version of making calls as well as an async version. Consider the synchronous way to fetch records from an SQS queue:

import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest
import software.amazon.awssdk.services.sqs.SqsClient

val receiveMessageRequest: ReceiveMessageRequest = ReceiveMessageRequest.builder()
    .queueUrl(queueUrl)
    .maxNumberOfMessages(5)
    .waitTimeSeconds(10)
    .build()

val messages: List<Message> = sqsClient.receiveMessage(receiveMessageRequest).messages()

Here "software.amazon.awssdk.services.sqs.SqsClient" is being used for querying sqs and retrieving a batch of results synchronously. An async result, on the other hand, looks like this:

val receiveMessageRequest: ReceiveMessageRequest = ReceiveMessageRequest.builder()
    .queueUrl(queueUrl)
    .maxNumberOfMessages(5)
    .waitTimeSeconds(10)
    .build()

val messages: CompletableFuture<List<Message>> = sqsAsyncClient
    .receiveMessage(receiveMessageRequest)
    .thenApply { result -> result.messages() }

The output now is now a "CompletableFuture"

Infinite loop and no backpressure

My first attempt at creating a stream(Flux) of message is fairly simple - an infinite loop that polls AWS sqs and creates a Flux from it using the "Flux.create" operator, this way:

fun listen(): Flux<Pair<String, () -> Unit>> {
    return Flux.create { sink: FluxSink<List<Message>> ->
            while (running) {
                try {
                    val receiveMessageRequest: ReceiveMessageRequest = ReceiveMessageRequest.builder()
                        .queueUrl(queueUrl)
                        .maxNumberOfMessages(5)
                        .waitTimeSeconds(10)
                        .build()

                    val messages: List<Message> = sqsClient.receiveMessage(receiveMessageRequest).messages()
                    LOGGER.info("Received: $messages")
                    sink.next(messages)
                } catch (e: InterruptedException) {
                    LOGGER.error(e.message, e)
                } catch (e: Exception) {
                    LOGGER.error(e.message, e)
                }
            }
        }
        .flatMapIterable(Function.identity())
        .doOnError { t: Throwable -> LOGGER.error(t.message, t) }
        .retry()
        .map { snsMessage: Message ->
            val snsMessageBody: String = snsMessage.body()
            val snsNotification: SnsNotification = readSnsNotification(snsMessageBody)
            snsNotification.message to { deleteQueueMessage(snsMessage.receiptHandle(), queueUrl) }
        }
}

The way this works is that there is an infinite loop that checks for new messages using long-polling. Messages may not be available at every poll, in which case an empty list is added to the stream.

This list of atmost 5 messages is then mapped to a stream of individual messages using the "flatMapIterable" operator, which is further mapped by extracting the message from the SNS wrapper (as message gets forwarded from SNS to SQS, SNS adds a wrapper to the message) and a way to delete the message(deleteHandle) once the message is successfully processed is returned as Pair.


This approach works perfectly fine... but imagine a case where a huge number of messages have come in, since the loop is not really aware of the throughput downstream it will keep pumping data to the stream. The default behavior is for the intermediate operators to buffer this data flowing in based on how the final consumer is consuming the data. Since this buffer is unbounded it is possible that the system may reach an unsustainable state.

Backpressure aware stream

The fix is to use a different operator to generate the stream of data - Flux.generate.
Using this operator the code looks like this:

fun listen(): Flux<Pair<String, () -> Unit>> {
    return Flux.generate { sink: SynchronousSink<List<Message>> ->
            val receiveMessageRequest: ReceiveMessageRequest = ReceiveMessageRequest.builder()
                .queueUrl(queueUrl)
                .maxNumberOfMessages(5)
                .waitTimeSeconds(10)
                .build()

            val messages: List<Message> = sqsClient.receiveMessage(receiveMessageRequest).messages()
            LOGGER.info("Received: $messages")
            sink.next(messages)
        }
        .flatMapIterable(Function.identity())
        .doOnError { t: Throwable -> LOGGER.error(t.message, t) }
        .retry()
        .map { snsMessage: Message ->
            val snsMessageBody: String = snsMessage.body()
            val snsNotification: SnsNotification = readSnsNotification(snsMessageBody)
            snsNotification.message to { deleteQueueMessage(snsMessage.receiptHandle(), queueUrl) }
        }
}

The way this works is that the block passed to the "Flux.generate" operator is repeatedly called - similar to the while loop, in each loop one item is expected to be added to the stream. In this instance, the item added to the stream happens to be a list which like before is broken down into individual messages.

How does backpressure work in this scenario -
So again consider the case where the downstream consumer is processing at a slower rate than the generating end. In this case, Flux itself would slow down at the rate at which the generate operator is called, thus being considerate of the throughput of the downstream system.

Conclusion

This should set up a good pipeline for processing messages from SQS, there are a few more nuances to this to process messages in parallel later in the stream which I will cover in a future post.

The codebase of this example is available in my github repository here - https://github.com/bijukunjummen/boot-with-sns-sqs. The code has a complete pipeline which includes processing the message and deleting it once processed.

Unit testing DynamoDB applications using JUnit5

In a previous post I had described the new AWS SDK for Java 2 which provides non-blocking IO support for Java clients calling different AWS services. In this post I will go over an approach that I have followed to unit test the AWS DynamoDB calls.

There are a few ways to spin up a local version of DynamoDB -

1. AWS provides a DynamoDB local
2. Localstack provides a way to spin up a good number of AWS services locally
3. A docker version of DynamoDB Local
4. Dynalite, a node based implementation of DynamoDB


Now to be able to unit test an application, I need to be able to start up an embedded version of DynamoDB using one of these options right before a test runs and then shut it down after a test completes. There are three approaches that I have taken:

1. Using a JUnit 5 extension that internally brings up a AWS DynamoDB Local and spins it down after a test.
2. Using testcontainers to start up a docker version DynamoDB Local
3. Using testcontainers to start up DynaLite

JUnit5 extension

JUnit5 extension provides a convenient hook point to start up an embedded version of DynamoDB for tests. It works by pulling in a version of DynamoDB Local as a maven dependency:

dependencies {
    ...
 testImplementation("com.amazonaws:DynamoDBLocal:1.11.119")
    ...
}

A complication with this dependency is that there are native components (dll, .so etc) that the DynamoDB Local interacts with and to get these in the right place, I depend on a Gradle task:

task copyNativeDeps(type: Copy) {
 mkdir "build/native-libs"
 from(configurations.testCompileClasspath) {
  include '*.dll'
  include '*.dylib'
  include '*.so'
 }
 into 'build/native-libs'
}

test {
 dependsOn copyNativeDeps
}

which puts the native libs in build/native-libs folder, and the extension internally sets this path as a system property:

System.setProperty("sqlite4java.library.path", libPath.toAbsolutePath().toString())

Here is the codebase to the JUnit5 extension with all these already hooked up - https://github.com/bijukunjummen/boot-with-dynamodb/blob/master/src/test/kotlin/sample/dyn/rules/LocalDynamoExtension.kt

A test using this extension looks like this:

class HotelRepoTest {
    companion object {
        @RegisterExtension
        @JvmField
        val localDynamoExtension = LocalDynamoExtension()

        @BeforeAll
        @JvmStatic
        fun beforeAll() {
            val dbMigrator = DbMigrator(localDynamoExtension.syncClient!!)
            dbMigrator.migrate()
        }

    }
    @Test
    fun saveHotel() {
        val hotelRepo = DynamoHotelRepo(localDynamoExtension.asyncClient!!)
        val hotel = Hotel(id = "1", name = "test hotel", address = "test address", state = "OR", zip = "zip")
        val resp = hotelRepo.saveHotel(hotel)

        StepVerifier.create(resp)
                .expectNext(hotel)
                .expectComplete()
                .verify()
    }
}

The code can interact with a fully featured DynamoDB.

TestContainers with DynamoDB Local Docker

The JUnit5 extensions approach works well but it requires an additional dependency with native binaries to be pulled in. A cleaner approach may be to use the excellent Testcontainers to spin up a docker version of DynamoDB Local the following way:

class HotelRepoLocalDynamoTestContainerTest {
    @Test
    fun saveHotel() {
        val hotelRepo = DynamoHotelRepo(getAsyncClient(dynamoDB))
        val hotel = Hotel(id = "1", name = "test hotel", address = "test address", state = "OR", zip = "zip")
        val resp = hotelRepo.saveHotel(hotel)

        StepVerifier.create(resp)
                .expectNext(hotel)
                .expectComplete()
                .verify()
    }



    companion object {
        val dynamoDB: KGenericContainer = KGenericContainer("amazon/dynamodb-local:1.11.119")
                .withExposedPorts(8000)

        @BeforeAll
        @JvmStatic
        fun beforeAll() {
            dynamoDB.start()
        }

        @AfterAll
        @JvmStatic
        fun afterAll() {
            dynamoDB.stop()
        }

        fun getAsyncClient(dynamoDB: KGenericContainer): DynamoDbAsyncClient {
            val endpointUri = "http://" + dynamoDB.getContainerIpAddress() + ":" +
                    dynamoDB.getMappedPort(8000)
            val builder: DynamoDbAsyncClientBuilder = DynamoDbAsyncClient.builder()
                    .endpointOverride(URI.create(endpointUri))
                    .region(Region.US_EAST_1)
                    .credentialsProvider(StaticCredentialsProvider
                            .create(AwsBasicCredentials
                                    .create("acc", "sec")))
            return builder.build()
        }

        ...
    }
}

This code starts up DynamoDB at a random unoccupied port and provides this information so that the client can be created using this information. There is a little Kotlin workaround that I had to do based on an issue reported here - https://github.com/testcontainers/testcontainers-java/issues/318

TestContainers with Dynalite

Dynalite is a javascript based implementation of DynamoDB and can be run for tests again using the TestContainer approach. This time however there is already a TestContainer module for Dynalite. I found that it does not support JUnit5 and sent a Pull request to provide this support, in the iterim the raw docker image can be used and this is how a test looks like:

class HotelRepoDynaliteTestContainerTest {
    @Test
    fun saveHotel() {
        val hotelRepo = DynamoHotelRepo(getAsyncClient(dynamoDB))
        val hotel = Hotel(id = "1", name = "test hotel", address = "test address", state = "OR", zip = "zip")
        val resp = hotelRepo.saveHotel(hotel)

        StepVerifier.create(resp)
                .expectNext(hotel)
                .expectComplete()
                .verify()
    }

    companion object {
        val dynamoDB: KGenericContainer = KGenericContainer("quay.io/testcontainers/dynalite:v1.2.1-1")
                .withExposedPorts(4567)

        @BeforeAll
        @JvmStatic
        fun beforeAll() {
            dynamoDB.start()
            val dbMigrator = DbMigrator(getSyncClient(dynamoDB))
            dbMigrator.migrate()
        }

        @AfterAll
        @JvmStatic
        fun afterAll() {
            dynamoDB.stop()
        }

        fun getAsyncClient(dynamoDB: KGenericContainer): DynamoDbAsyncClient {
            val endpointUri = "http://" + dynamoDB.getContainerIpAddress() + ":" +
                    dynamoDB.getMappedPort(4567)
            val builder: DynamoDbAsyncClientBuilder = DynamoDbAsyncClient.builder()
                    .endpointOverride(URI.create(endpointUri))
                    .region(Region.US_EAST_1)
                    .credentialsProvider(StaticCredentialsProvider
                            .create(AwsBasicCredentials
                                    .create("acc", "sec")))
            return builder.build()
        }
        ...
    }
}

Conclusion

All of the approaches are useful in being able to test integration with DynamoDB. My personal preference is using the TestContainers approach if a docker agent is available else with the JUnit5 extension approach. The samples with fully working tests using all the three approaches are available in my github repo - https://github.com/bijukunjummen/boot-with-dynamodb