Tuesday, August 6, 2019

Chicken and egg - resolving Spring properties ahead of a test

Consider a service class responsible for making a remote call and retrieving a detail:


...
public class CitiesService {
    private final WebClient.Builder webClientBuilder;
    private final String baseUrl;

    public CitiesService(
            WebClient.Builder webClientBuilder,
            @Value("${cityservice.url}") String baseUrl) {
        this.webClientBuilder = webClientBuilder;
        this.baseUrl = baseUrl;
    }


    public Flux<City> getCities() {
        return this.webClientBuilder.build()
                .get()
....

This is a Spring Bean and resolves the url to call through a property called "cityservice.url".

If I wanted to test this class, an approach that I have been using when using WebClient is to start a mock server using the excellent Wiremock and using it to test this class. A Wiremock mock looks like this:

    private static final WireMockServer WIREMOCK_SERVER =
            new WireMockServer(wireMockConfig().dynamicPort());


    .....

    WIREMOCK_SERVER.stubFor(get(urlEqualTo("/cities"))
                .withHeader("Accept", equalTo("application/json"))
                .willReturn(aResponse()
                        .withStatus(200)
                        .withHeader("Content-Type", "application/json")
                        .withBody(resultJson)));

The Wiremock server is being started up at a random port and it is set to respond to an endpoint called "/cities". Here is where the chicken and egg problem comes up:

1. The CitiesService class requires a property called "cityservice.url" to be set before starting the test.
2. Wiremock is started at a random port and the url that it is responding to is "http://localhost:randomport" and is available only once the test is kicked off.


There are three potential solutions that I can think of to break this circular dependency:

Approach 1: To use a hardcoded port

This approach depends on starting up Wiremock on a fixed port instead of a dynamic port, this way the property can be set when starting the test up, something like this:

@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = CitiesServiceHardcodedPortTest.SpringConfig.class,
        properties = "cityservice.url=http://localhost:9876")
public class CitiesServiceHardcodedPortTest {
    private static final WireMockServer WIREMOCK_SERVER =
            new WireMockServer(wireMockConfig().port(9876));


Here Wiremock is being started at port 9876 and the property at startup is being set to "http://localhost:9876/".

This solves the problem, however, this is not CI server friendly, it is possible for the ports to collide at runtime and this makes for a flaky test.


Approach 2: Not use Spring for test

A better approach is to not use the property, along these lines:

public class CitiesServiceDirectTest {
    private static final WireMockServer WIREMOCK_SERVER =
            new WireMockServer(wireMockConfig().dynamicPort());

    private CitiesService citiesService;

    @BeforeEach
    public void beforeEachTest() {
        final WebClient.Builder webClientBuilder = WebClient.builder();

        this.citiesService = new CitiesService(webClientBuilder, WIREMOCK_SERVER.baseUrl());
    }


Here the service is being created by explicitly setting the baseUrl in the constructor, thus avoiding the need to set a property ahead of the test.


Approach 3: Application Context Initializer


ApplicationContextInitializer is used for programmatically initializing a Spring Application Context and it can be used with a test to inject in the property before the actual test is executed. Along these lines:

@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = CitiesServiceSpringTest.SpringConfig.class)
@ContextConfiguration(initializers = {CitiesServiceSpringTest.PropertiesInitializer.class})
public class CitiesServiceSpringTest {
    private static final WireMockServer WIREMOCK_SERVER =
            new WireMockServer(wireMockConfig().dynamicPort());

    @Autowired
    private CitiesService citiesService;

    @Test
    public void testGetCitiesCleanFlow() throws Exception {
        ...
    }



    static class PropertiesInitializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {

        @Override
        public void initialize(ConfigurableApplicationContext applicationContext) {
            TestPropertyValues.of(
                    "cityservice.url=" + "http://localhost:" + WIREMOCK_SERVER.port()
            ).applyTo(applicationContext.getEnvironment());
        }
    }

}

Wiremock is started up first, then Spring context is initialized using the initializer which injects in the "cityservice.url" property using the Wiremocks dynamic port, this way the property is available for wiring into CityService.


Conclusion

I personally prefer Approach 2, however it is good to have Spring's wiring and the dependent beans created ahead of the test and if the class utilizes these then I prefer Approach 3. Application Context initializer provides a good way to break the chicken and egg problem with properties like these which need to be available ahead of Spring's context getting engaged.

All the code samples are available here:

Approach 1: https://github.com/bijukunjummen/reactive-cities-demo/blob/master/src/test/java/samples/geo/service/CitiesServiceHardcodedPortTest.java
Approach 2: https://github.com/bijukunjummen/reactive-cities-demo/blob/master/src/test/java/samples/geo/service/CitiesServiceDirectTest.java
Approach 3: https://github.com/bijukunjummen/reactive-cities-demo/blob/master/src/test/java/samples/geo/service/CitiesServiceSpringTest.java

Sunday, June 9, 2019

Callback hell and Reactive patterns

One of the ways that I have better understood the usefulness of a Reactive Streams based approach is how it simplifies a Non-blocking IO call.

This post will be a quick walkthrough of the kind of code involved in making a synchronous remote call, then show how layering in Non-blocking IO though highly efficient in the use of resources(especially threads) introduces complications referred to as a callback hell and how a reactive streams based approach simplifies the programming model.


Target Service


Since I will be writing a client call, my target service representing the details of a City has two endpoints. One returning a list of city id's when called with a uri of type - "/cityids" and a sample result looks like this:

[
    1,
    2,
    3,
    4,
    5,
    6,
    7
]

and an endpoint returning the details of a city given its id, for example when called using an id of 1 - "/cities/1":

{
    "country": "USA",
    "id": 1,
    "name": "Portland",
    "pop": 1600000
}

The client's responsibility is to get the list of city id's and then for each city id get the detail of the city and put it together into a list of cities.

Synchronous call


I am using Spring Framework's RestTemplate to make the remote call. A Kotlin function to get the list of cityids looks like this:

private fun getCityIds(): List<String> {
    val cityIdsEntity: ResponseEntity<List<String>> = restTemplate
            .exchange("http://localhost:$localServerPort/cityids",
                    HttpMethod.GET,
                    null,
                    object : ParameterizedTypeReference<List<String>>() {})
    return cityIdsEntity.body!!
}

and to get the details of a city:

private fun getCityForId(id: String): City {
    return restTemplate.getForObject("http://localhost:$localServerPort/cities/$id", City::class.java)!!
}

Given these two functions it is easy to compose them such that a list of cities is returned:

val cityIds: List<String> = getCityIds()
val cities: List<City> = cityIds
        .stream()
        .map<City> { cityId -> getCityForId(cityId) }
        .collect(Collectors.toList())

cities.forEach { city -> LOGGER.info(city.toString()) }

The code is very easy to understand, however, there are 8 blocking calls involved -
1. to get the list of 7 city ids and then to get the details for each
2. To get the details of each of the 7 cities

Each of these calls would have been on a different thread.

Using Non-Blocking IO with callback

I will be using a library called AsyncHttpClient to make a non-blocking IO call.

AyncHttpClient returns a ListenableFuture type when a remote call is made.

val responseListenableFuture: ListenableFuture<Response> = asyncHttpClient
                .prepareGet("http://localhost:$localServerPort/cityids")
                .execute()

A callback can be attached to a Listenable future to act on the response when available.

responseListenableFuture.addListener(Runnable {
    val response: Response = responseListenableFuture.get()
    val responseBody: String = response.responseBody
    val cityIds: List<Long> = objectMapper.readValue<List<Long>>(responseBody,
            object : TypeReference<List<Long>>() {})
    ....
}        

Given the list of cityids I want to get the details of the city, so from the response I need to make more remote calls and attach a callback for each of the calls to get the details of the city along these lines:

val responseListenableFuture: ListenableFuture<Response> = asyncHttpClient
        .prepareGet("http://localhost:$localServerPort/cityids")
        .execute()

responseListenableFuture.addListener(Runnable {
    val response: Response = responseListenableFuture.get()
    val responseBody: String = response.responseBody
    val cityIds: List<Long> = objectMapper.readValue<List<Long>>(responseBody,
            object : TypeReference<List<Long>>() {})

    cityIds.stream().map { cityId ->
        val cityListenableFuture = asyncHttpClient
                .prepareGet("http://localhost:$localServerPort/cities/$cityId")
                .execute()

        cityListenableFuture.addListener(Runnable {
            val cityDescResp = cityListenableFuture.get()
            val cityDesc = cityDescResp.responseBody
            val city = objectMapper.readValue(cityDesc, City::class.java)
            LOGGER.info("Got city: $city")
        }, executor)
    }.collect(Collectors.toList())
}, executor)

This is a gnarly piece of code, there is set of callbacks within a callback which is very difficult to reason about and make sense of and hence referred to as the callback hell.


Using Non-Blocking IO with Java CompletableFuture


This code can be improved a little by returning a Java's CompletableFuture as the return type instead of the ListenableFuture. CompletableFuture provides operators that allow the return type to modified and returned.

As an example, consider the function to get the list of city ids:

private fun getCityIds(): CompletableFuture<List<Long>> {
    return asyncHttpClient
            .prepareGet("http://localhost:$localServerPort/cityids")
            .execute()
            .toCompletableFuture()
            .thenApply { response ->
                val s = response.responseBody
                val l: List<Long> = objectMapper.readValue(s, object : TypeReference<List<Long>>() {})
                l
            }
}

Here I am using the "thenApply" operator to transform "CompletableFuture<Response>" to "CompletableFuture<List<Long>>

And similarly to get the detail a city:

private fun getCityDetail(cityId: Long): CompletableFuture<City> {
    return asyncHttpClient.prepareGet("http://localhost:$localServerPort/cities/$cityId")
            .execute()
            .toCompletableFuture()
            .thenApply { response ->
                val s = response.responseBody
                LOGGER.info("Got {}", s)
                val city = objectMapper.readValue(s, City::class.java)
                city
            }
}

This is an improvement from the Callback based approach, however, CompletableFuture lacks sufficient operators, say in this specific instance where all the city details need to be put together:

val cityIdsFuture: CompletableFuture<List<Long>> = getCityIds()
val citiesCompletableFuture: CompletableFuture<List<City>> =
        cityIdsFuture
                .thenCompose { l ->
                    val citiesCompletable: List<CompletableFuture<City>> =
                            l.stream()
                                    .map { cityId ->
                                        getCityDetail(cityId)
                                    }.collect(toList())

                    val citiesCompletableFutureOfList: CompletableFuture<List<City>> =
                            CompletableFuture.allOf(*citiesCompletable.toTypedArray())
                                    .thenApply { _: Void? ->
                                        citiesCompletable
                                                .stream()
                                                .map { it.join() }
                                                .collect(toList())
                                    }
                    citiesCompletableFutureOfList
                }

I have used an operator called CompletableFuture.allOf which returns a "Void" type and has to be coerced to return the desired type of ""CompletableFuture<List<City>>.


Using Project Reactor

Project Reactor is an implementation of the Reactive Streams specification. It has two specialized types to return a stream of 0/1 item and a stream of 0/n items - the former is a Mono, the latter a Flux.

Project Reactor provides a very rich set of operators that allow the stream of data to be transformed in a variety of ways. Consider first the function to return a list of City ids:

private fun getCityIds(): Flux<Long> {
    return webClient.get()
            .uri("/cityids")
            .exchange()
            .flatMapMany { response ->
                LOGGER.info("Received cities..")
                response.bodyToFlux<Long>()
            }
}

I am using Spring's excellent WebClient library to make the remote call and get a Project reactor "Mono<ClientResponse>" type of response, which can be modified to a "Flux<Long>" type using the "flatMapMany" operator.

Along the same lines to get the detail of the city, given a city id:

private fun getCityDetail(cityId: Long?): Mono<City> {
    return webClient.get()
            .uri("/cities/{id}", cityId!!)
            .exchange()
            .flatMap { response ->
                val city: Mono<City> = response.bodyToMono()
                LOGGER.info("Received city..")
                city
            }
}

Here a Project reactor "Mono<ClientResponse>" type is being transformed to "Mono<City>" type using the "flatMap" operator.

and the code to get the cityids and then the City's from it:

val cityIdsFlux: Flux<Long> = getCityIds()
val citiesFlux: Flux<City> = cityIdsFlux
        .flatMap { this.getCityDetail(it) }

return citiesFlux


This is very expressive - contrast the mess of a callback based approach and the simplicity of the reactive streams based approach.


Conclusion

In my mind, this is one of the biggest reasons to use a Reactive Streams based approach and in particular Project Reactor for scenarios that involve crossing asynchronous boundaries like in this instance to make remote calls. It cleans up the mess of callbacks and callback hells and provides a natural approach of modifying/transforming types using a rich set of operators.


My repository with a working version of all the samples that I have used here is available at https://github.com/bijukunjummen/reactive-cities-demo/tree/master/src/test/kotlin/samples/geo/kotlin

Tuesday, April 30, 2019

Functional Hystrix using Spring Cloud HystrixCommands

Spring's WebClient provides a non-blocking client for making service to service calls. Hystrix, though now in a maintenance mode, has been used for protecting service to service calls by preventing cascading failures, providing circuit breakers for calls to slow or faulty upstream services.

In this post, I will be exploring how Spring Cloud provides a newer functional approach to wrapping a remote call with Hystrix.


Consider a simple service that returns a list of entities, say a list of cities, modeled using the excellent Wiremock tool:

WIREMOCK_SERVER.stubFor(WireMock.get(WireMock.urlMatching("/cities"))
                .withHeader("Accept", WireMock.equalTo("application/json"))
                .willReturn(WireMock.aResponse()
                        .withStatus(HttpStatus.OK.value())
                        .withFixedDelay(5000)
                        .withHeader("Content-Type", "application/json")))

When called with a uri of the type "/cities" this Wiremock endpoint responds with a json of the following type:

[
  {
    "country": "USA",
    "id": 1,
    "name": "Portland",
    "pop": 1600000
  },
  {
    "country": "USA",
    "id": 2,
    "name": "Seattle",
    "pop": 3200000
  },
  {
    "country": "USA",
    "id": 3,
    "name": "SFO",
    "pop": 6400000
  }
]

after a delay of 5 seconds.

Traditional approach


There are many approaches to using Hystrix, I have traditionally preferred an approach where an explicit Hystrix Command protects the remote call, along these lines:

import com.netflix.hystrix.HystrixCommandGroupKey
import com.netflix.hystrix.HystrixCommandKey
import com.netflix.hystrix.HystrixCommandProperties
import com.netflix.hystrix.HystrixObservableCommand
import org.bk.samples.model.City
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.http.MediaType
import org.springframework.web.reactive.function.client.WebClient
import org.springframework.web.reactive.function.client.bodyToFlux
import org.springframework.web.util.UriComponentsBuilder
import reactor.core.publisher.Flux
import rx.Observable
import rx.RxReactiveStreams
import rx.schedulers.Schedulers
import java.net.URI


class CitiesHystrixCommand(
        private val webClientBuilder: WebClient.Builder,
        private val citiesBaseUrl: String
) : HystrixObservableCommand<City>(
        HystrixObservableCommand.Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("cities-service"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("cities-service"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionTimeoutInMilliseconds(4000))) {
    override fun construct(): Observable<City> {
        val buildUri: URI = UriComponentsBuilder
                .fromUriString(citiesBaseUrl)
                .path("/cities")
                .build()
                .encode()
                .toUri()

        val webClient: WebClient = this.webClientBuilder.build()

        val result: Flux<City> = webClient.get()
                .uri(buildUri)
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .flatMapMany { clientResponse ->
                    clientResponse.bodyToFlux<City>()
                }

        return RxReactiveStreams.toObservable(result)
    }

    override fun resumeWithFallback(): Observable<City> {
        LOGGER.error("Falling back on cities call", executionException)
        return Observable.empty()
    }

    companion object {
        private val LOGGER: Logger = LoggerFactory.getLogger(CitiesHystrixCommand::class.java)
    }
}

This code can now be used to make a remote call the following way:

import org.springframework.http.MediaType
import org.springframework.web.reactive.function.client.WebClient


class CitiesHystrixCommandBasedClient(
        private val webClientBuilder: WebClient.Builder,
        private val citiesBaseUrl: String
) {
    fun getCities(): Flux<City> {
        val citiesObservable: Observable<City> = CitiesHystrixCommand(webClientBuilder, citiesBaseUrl)
                .observe()
                .subscribeOn(Schedulers.io())

        return Flux
                .from(RxReactiveStreams
                        .toPublisher(citiesObservable))
    }
}


Two things to note here,
1. WebClient returns a Project Reactor "Flux" type representing a list of cities, however Hystrix is Rx-Java 1 based, so Flux is being transformed to Rx-Java Observable using "RxReactiveStreams.toObservable()" call, provided by the RxJavaReactiveStreams library here.

2. I still want Project Reactor "Flux" type to be used in the rest of the application, so there is another adapter that converts the Rx-Java Observable back to a Flux - "Flux.from(RxReactiveStreams.toPublisher(citiesObservable))" once the call wrapped in Hystrix returns.


If I were to try this client with the wiremock sample with the 5 second delay, it correctly handles the delay and returns after a second.


Functional approach


There is a lot of boiler-plate with the previous approach which is avoided with the new functional approach of using HystrixCommands, a utility class which comes with Spring Cloud which provides a functional approach to making the remote call wrapped with Hystrix.

The entirety of the call using HystrixCommands looks like this:

import com.netflix.hystrix.HystrixCommandProperties
import org.bk.samples.model.City
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.cloud.netflix.hystrix.HystrixCommands
import org.springframework.http.MediaType
import org.springframework.web.reactive.function.client.WebClient
import org.springframework.web.reactive.function.client.bodyToFlux
import org.springframework.web.util.UriComponentsBuilder
import reactor.core.publisher.Flux
import rx.schedulers.Schedulers
import java.net.URI

class CitiesFunctionalHystrixClient(
        private val webClientBuilder: WebClient.Builder,
        private val citiesBaseUrl: String
) {
    fun getCities(): Flux<City> {
        return HystrixCommands
                .from(callCitiesService())
                .commandName("cities-service")
                .groupName("cities-service")
                .commandProperties(
                        HystrixCommandProperties.Setter()
                                .withExecutionTimeoutInMilliseconds(1000)
                )
                .toObservable { obs ->
                    obs.observe()
                            .subscribeOn(Schedulers.io())
                }
                .fallback { t: Throwable ->
                    LOGGER.error(t.message, t)
                    Flux.empty()
                }
                .toFlux()
    }

    fun callCitiesService(): Flux<City> {
        val buildUri: URI = UriComponentsBuilder
                .fromUriString(citiesBaseUrl)
                .path("/cities")
                .build()
                .encode()
                .toUri()

        val webClient: WebClient = this.webClientBuilder.build()

        return webClient.get()
                .uri(buildUri)
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .flatMapMany { clientResponse ->
                    clientResponse.bodyToFlux<City>()
                }
    }

    companion object {
        private val LOGGER: Logger = LoggerFactory.getLogger(CitiesHystrixCommand::class.java)
    }
}


A lot of boiler-plate is avoided with this approach -
1. an explicit command is not required anymore
2. the call and the fallback are coded in a fluent manner
3. Any overrides can be explicitly specified - in this specific instance the timeout of 1 second.

Conclusion

I like the conciseness which HystrixCommands brings to the usage of Hystrix with WebClient. I have the entire sample available in my github repo - https://github.com/bijukunjummen/webclient-hystrix-sample, all the dependencies required to get the samples to work is part of this repo. If you are interested in sticking with Rx-Java 1, then an approach described here may help you avoid boiler-plate with vanilla Hystrix


Tuesday, March 5, 2019

Water Pouring Problem with Kotlin and Vavr

The first time I saw the Water Pouring Problem being programmatically solved was the excellent lectures on functional Programming by Martin Odersky on Coursera. The solution demonstrates the power of lazy evaluation in Streams with Scala.

Solving Water Pouring Problem using Kotlin

I wanted to explore how I can rewrite the solution described by Martin Odersky using Kotlin and I realized two things - one is that the immutable data structures that Kotlin offers are simply wrappers over Java Collections library and are not truly immutable, secondly the solution using Streams feature in Java will be difficult. However, the Vavr offers a good alternative to both - a first-class Immutable collections library and a Streams library and I took a crack at replicating the solution with Kotlin and Vavr.

A Cup looks like this, represented as a Kotlin data class:

import io.vavr.collection.List


data class Cup(val level: Int, val capacity: Int) {
    override fun toString(): String {
        return "Cup($level/$capacity)"
    }
}

Since the water pouring problem repesents the "state" of a set of cups, this can be simply represented as a "typealias" the following way:
typealias State = List<Cup>


There are 3 different types of moves that can be performed with the water in the cup - Empty it, Fill it, or Pour from one cup to another, represented again as Kotlin Data classes:

interface Move {
    fun change(state: State): State
}

data class Empty(val glass: Int) : Move {
    override fun change(state: State): State {
        val cup = state[glass]
        return state.update(glass, cup.copy(level = 0))
    }

    override fun toString(): String {
        return "Empty($glass)"
    }
}

data class Fill(val glass: Int) : Move {
    override fun change(state: State): State {
        val cup = state[glass]
        return state.update(glass, cup.copy(level = cup.capacity))
    }

    override fun toString(): String {
        return "Fill($glass)"
    }
}

data class Pour(val from: Int, val to: Int) : Move {
    override fun change(state: State): State {
        val cupFrom = state[from]
        val cupTo = state[to]
        val amount = min(cupFrom.level, cupTo.capacity - cupTo.level)

        return state
                .update(from, cupFrom.copy(cupFrom.level - amount))
                .update(to, cupTo.copy(level = cupTo.level + amount))
    }

    override fun toString(): String {
        return "Pour($from,$to)"
    }
}

The implementation is making use of Vavr's List data structures "update" method to create a new list with just the relevant elements updated.


A "Path" represents a history of moves leading to the current state:

data class Path(val initialState: pour.State, val endState: State, val history: List<Move>) {
    fun extend(move: Move) = Path(initialState, move.change(endState), history.prepend(move))

    override fun toString(): String {
        return history.reverse().mkString(" ") + " ---> " + endState
    }
}

I am using the "prepend" method of the list to add elements to the beginning of history. Prepending to a list is an O(1) operation whereas appending is O(n), hence the choice.

Given a "state", a set of possible moves to change the "state" are the following -

1. Empty the glasses -

(0 until count).map { Empty(it) }

2. Fill the glasses -

(0 until count).map { Fill(it) }

3. Pour from one glass to another -

(0 until count).flatMap { from ->
    (0 until initialState.length()).filter { to -> from != to }.map { to ->
        Pour(from, to)
    }
}

Now, all these moves are used for advancing from one state to another. Consider say 2 cups with capacity of 4 and 9 litres, initially filled with 0 litres of water, represented as "List(Cup(0/4), Cup(0/9))", with all possible moves the next set of states of the cups are the following:


Similarly, advancing each of these states to a new set of states would like this(in a somewhat simplified form):


As each State advances to a next set of states based on all possible moves, it can be seen that there will be an explosion of possible paths, this is where laziness offered by the Stream data structure of Vavr comes in. The values in a stream are only computed on request.


Given a set of paths, new paths are created using Stream the following way:

fun from(paths: Set<Path>, explored: Set<State>): Stream<Set<Path>> {
    if (paths.isEmpty) {
        return Stream.empty()
    } else {
        val more = paths.flatMap { path ->
            moves.map { move ->
                val next: Path = path.extend(move)
                next
            }.filter { !explored.contains(it.endState) }
        }
        return Stream.cons(paths) { from(more, explored.addAll(more.map { it.endState })) }
    }
}


So, now given a stream of potential paths from the initial state to a new state, a solution to a "target" state becomes:

val pathSets = from(hashSet(initialPath), hashSet())

fun solution(target: State): Stream<Path> {
    return pathSets.flatMap { it }.filter { path -> path.endState == target }
}



That covers the solution, a test with this code looks like this - there are two cups of 4 litre and 9 litre capacity, initially filled with 0 litres of water. The final target state is to get the second cup filled with 6 litres of water:

val initialState = list(Cup(0, 4), Cup(0, 9))
val pouring = Pouring(initialState)

pouring.solution(list(Cup(0, 4), Cup(6, 9)))
    .take(1).forEach { path ->
        println(path)
    }

when run, this spits out the following solution:

Fill(1) Pour(1,0) Empty(0) Pour(1,0) Empty(0) Pour(1,0) Fill(1) Pour(1,0) Empty(0) ---> List(Cup(0/4), Cup(6/9))

Graphically represented, the solution looks like this:



It may be easier to simply follow a working version of the sample which is available in my github repo - https://github.com/bijukunjummen/algos/blob/master/src/test/kotlin/pour/Pouring.kt


Conclusion

Although Kotlin lacks first class support for native immutable datastructures, I feel that a combination of Vavr with Kotlin makes for a solution that is as elegant as the Scala one.

Monday, January 21, 2019

Spring-Boot 2.1.x and overriding bean definition

I was recently migrating an application from Spring Boot 1.5.X to Spring Boot 2.X and saw an issue with overriding Spring Bean definitions. One of the configurations was along these lines in Kotlin:


@Configuration
class DynamoConfig {

    @Bean
    fun dynamoDbAsyncClient(dynamoProperties: DynamoProperties): DynamoDbAsyncClient {
        ...
    }

    @Bean
    fun dynampoDbSyncClient(dynamoProperties: DynamoProperties): DynamoDbClient {
        ...
    }
}


Now, for a test I wanted to override these 2 bean definitions and did something along these lines:

@SpringBootTest
class DynamoConfigTest {

    @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()
    }


    @TestConfiguration
    class SpringConfig {
        @Bean
        fun dynamoDbAsyncClient(dynamoProperties: DynamoProperties): DynamoDbAsyncClient {
            ...
        }

        @Bean
        fun dynamoDbSyncClient(dynamoProperties: DynamoProperties): DynamoDbClient {
            ...
        }
    }
}

This type of overriding works with Spring Boot 1.5.X but fails with Spring Boot 2.1.X with an error:

Invalid bean definition with name 'dynamoDbAsyncClient' defined in sample.dyn.repo.DynamoConfigTest$SpringConfig:.. 
There is already .. defined in class path resource [sample/dyn/config/DynamoConfig.class]] bound



I feel this behavior is right, not allowing beans to overridden this way is the correct default behavior for an application, however I do want the ability to override the beans for tests and thanks to a Stack Overflow answer and Spring Boot 2.1.X release notes, the fix is to allow overrides using a property "spring.main.allow-bean-definition-overriding=true", so with this change, the test looks like this:

@SpringBootTest(properties = ["spring.main.allow-bean-definition-overriding=true"])
class DynamoConfigTest {

    @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()
    }


    @TestConfiguration
    class SpringConfig {
        @Bean
        fun dynamoDbAsyncClient(dynamoProperties: DynamoProperties): DynamoDbAsyncClient {
            ...
        }

        @Bean
        fun dynamoDbSyncClient(dynamoProperties: DynamoProperties): DynamoDbClient {
            ...
        }
    }
}

Friday, January 4, 2019

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

Sunday, December 16, 2018

Reactive Spring Webflux with AWS DynamoDB

AWS has released AWS SDK for Java version 2, the SDK now supports non-blocking IO for the API calls of different AWS services. In this post I will be exploring using the DynamoDB API's of the AWS SDK 2.x and using Spring Webflux stack to expose a reactive endpoint - this way the application is reactive end to end and presumably should use resources very efficiently (I have plans to do some tests on this set-up as a follow up).


Details of the Application


It may be easier to simply look at the code and follow it there - it is available in my GitHub repo.

The application is a simple one - to perform CRUD operation on a Hotel entity represented using the following Kotlin code:

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

I want to expose endpoints to save and retrieve a hotel entity and to get the list of hotels by state.


Details of the AWS SDK 2


The package names of the AWS SDK 2 api's all start with "software.amazon.awssdk" prefix now, the client to interact with DynamoDB is created using code along these lines:

import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient

val client: DynamoDbAsyncClient = DynamoDbAsyncClient.builder()
        .region(Region.of(dynamoProperties.region))
        .credentialsProvider(DefaultCredentialsProvider.builder().build())
        .build()


Once the DynamoDbAsyncClient instance is created, any operation using this client returns a Java 8 CompletableFuture type. For eg. in saving a Hotel entity:

val putItemRequest = PutItemRequest.builder()
        .tableName("hotels")
        .item(HotelMapper.toMap(hotel))
        .build()
        
val result: CompletableFuture<PutItemResponse> =
        dynamoClient.putItem(putItemRequest)

and in retrieving a record by id:

val getItemRequest: GetItemRequest = GetItemRequest.builder()
        .key(mapOf(Constants.ID to AttributeValue.builder().s(id).build()))
        .tableName(Constants.TABLE_NAME)
        .build()

val response: CompletableFuture<GetItemResponse> = dynamoClient.getItem(getItemRequest)

CompletableFuture provides a comprehensive set of functions to transform the results when available.

Integrating with Spring Webflux

Spring Webflux is a reactive web framework. The non-blocking IO support in AWS SDK 2 now makes it possible to write an end to end reactive and non-blocking applications with DynamoDB. Spring Webflux uses reactor-core to provide reactive streams support and the trick to integrating with AWS SDK 2 is to transform the Java 8 CompletableFuture to a reactor-core type, the following way when retrieving an item from DynamoDB by id:

val getItemRequest: GetItemRequest = GetItemRequest.builder()
        .key(mapOf(Constants.ID to AttributeValue.builder().s(id).build()))
        .tableName(Constants.TABLE_NAME)
        .build()

return Mono.fromCompletionStage(dynamoClient.getItem(getItemRequest))
        .map { resp ->
            HotelMapper.fromMap(id, resp.item())
        }

Spring Webflux expects the return types of the different web endpoint method signatures to be of reactive types, so a typical endpoint for getting say a list of hotels is the following:

@RequestMapping(value = ["/hotels"], method = [RequestMethod.GET])
fun getHotelsByState(@RequestParam("state") state: String): Flux<Hotel> {
    return hotelRepo.findHotelsByState(state)
}

Spring Webflux also supports a functional way to describe the API of the application, so an equivalent API to retrieve a hotel by its id, but expressed as a functional DSL is the following:

@Configuration
class HotelAdditionalRoutes {

    @Bean
    fun routes(hotelRepo: HotelRepo) = router {
        GET("/hotels/{id}") { req ->
            val id = req.pathVariable("id")
            val response: Mono<ServerResponse> = hotelRepo.getHotel(id)
                    .flatMap { hotel ->
                        ServerResponse.ok().body(BodyInserters.fromObject(hotel))
                    }
            response.switchIfEmpty(ServerResponse.notFound().build())
        }
    }
}


Conclusion

AWS SDK 2 makes it simple to write an end to end reactive and non-blocking applications. I have used Spring Webflux and AWS SDK 2 dynamo client to write such an application here. The entire working sample is available in my GitHub repo - https://github.com/bijukunjummen/boot-with-dynamodb, and has instructions on how to start up a local version of DynamoDB and use it for testing the application.