Reactive Microservices (Part 3)
This is part 3 of a series on reactive microservices, the first two parts are available at https://marlo.com.au/reactive-microservices and https://marlo.com.au/reactive-microservices-part-2
Photo by Mika Baumeister on Unsplash
At the end of the last post, I introduced the "Customer Microservice" from CompuHyperGlobalMegaNet1. The business at CompuHyperGlobalMegaNet have determined that a key plank in their digital strategy is to know who their customers are, so we have been asked to come up with something that will store the customer details. "Doesn’t Salesforce do that really well?" asked a junior dev. The CTO had a quick look at the web page, but to be honest it was pretty confusing and seemed complicated, so instead we decided to develop our own Customer microservice using Spring Boot.
We will build it old school first using Spring Web, then using Reactive to spell out the differences between the 2 approaches.
I’ve created the application scaffolding using Spring Initializr which is a really fast and simple way to get a project setup.
The Architect provided us with a spec:
And we got building: Spring Boot, REST API, JPA with Spring Data. Go2!
We defined a Customer class that can be saved into a database using JPA, and sprinkle some lombok on so we can replace code with annotations:-
@Entity
@NoArgsConstructor
@AllArgsConstructor
@Data
public class Customer {
@Id @GeneratedValue
private Integer id;
private String firstName;
private String lastName;
}More annotations than code, but Awesome.
Let’s create a DAO so we can database it (I know we say repositories these days, but it’s hard to change):
@EnableJpaRepositories
public interface CustomerDAO extends JpaRepository<Customer, Integer> {
}Spring Data is magic.
And finally a CustomerController to do the web things:
@RestController
@RequiredArgsConstructor
@Slf4j
public class CustomerController {
private final CustomerDAO customerDAO;
@GetMapping
public List<Customer> customers() {
return this.customerDAO.findAll();
}
}This starts, but is hard to test as there’s no data in the DB and no way to create any! We are going to need a "add customer" feature. Until we get that, let’s add an event to populate some customer data:
I am shamelessly stealing this trick (and quite a few reactive idioms further along as well) from Spring Legend Josh Long. Check out his talks and articles, I certainly have learned a LOT from them.
@RequiredArgsConstructor
@Component
class Initialise implements ApplicationListener<ApplicationReadyEvent> {
private final CustomerDAO customerDAO;
@Override
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
List.of("Brock:Mills","Brian:Fitzgerald").stream().map(s -> {
var c = s.split(":");
return new Customer(null, c[0], c[1], null, null);
}).forEach(c -> customerDAO.save(c));
customerDAO.findAll().stream().forEach(System.out::println);
}
}and we should probably do that addCustomer endpoint as well, so back to the CustomerController
@PostMapping
public ResponseEntity<Void> addCustomer(@RequestBody Customer customer) throws Exception {
this.customerDAO.save(customer);
return ResponseEntity.created(new URI("/" + customer.getId().toString())).build();
}Start it and test:
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.3.1.RELEASE)
2020-07-06 14:52:57.929 INFO 73930 --- [ main] c.m.n.NotreactiveDbApplication : Starting NotreactiveDbApplication on emmet.localdomain with PID 73930 (/Users/brockmills/Development/Marlo/microservices/notreactive-db/target/classes started by brockmills in /Users/brockmills/Development/Marlo/microservices/notreactive-db)
2020-07-06 14:52:57.933 INFO 73930 --- [ main] c.m.n.NotreactiveDbApplication : No active profile set, falling back to default profiles: default
2020-07-06 14:52:59.450 INFO 73930 --- [ main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data JPA repositories in DEFERRED mode.
2020-07-06 14:52:59.547 INFO 73930 --- [ main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 80ms. Found 1 JPA repository interfaces.
2020-07-06 14:53:00.259 INFO 73930 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 8080 (http)
2020-07-06 14:53:00.284 INFO 73930 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat]
... snip ...
2020-07-06 14:53:01.616 INFO 73930 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 8080 (http) with context path ''
... snip ...
2020-07-06 14:53:02.496 INFO 73930 --- [ main] DeferredRepositoryInitializationListener : Spring Data repositories initialized!
2020-07-06 14:53:02.509 INFO 73930 --- [ main] c.m.n.NotreactiveDbApplication : Started NotreactiveDbApplication in 5.405 seconds (JVM running for 6.239)
Customer(id=1, firstName=Brock, lastName=Mills)
Customer(id=2, firstName=Brian, lastName=Fitzgerald)Great, we are running with some data. Let’s test the API:
brockmills@emmet microservices % curl http://localhost:8080/
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"}]%Win! Let’s add a customer with a POST and query again:
brockmills@emmet microservices % curl --header 'Content-Type: application/json' --data '{"firstName":"Tester", "lastName":"Testing"}' http://localhost:8080/
brockmills@emmet microservices % curl http://localhost:8080/
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"},{"id":3,"firstName":"Tester","lastName":"Testing"}]%Ship it! I did hear the architect say something after a long lunch about unfunctionals? defunctionals? IDK. The new dev mentioned they thought security, error handling and data integrity were important, so we created some JIRAs for our tech debt sprint. We’ve got features to chuck over the fence at ops deliver!
While we were deploying to Prod, the business peeps dropped some new requirements on us. Apparently they have some amazing big data science lake thing going and MUST know the weather in Perth when the customer is created.
We had some questions about this – Why Perth? Which Perth? What actual thing about the weather? Wouldn’t it be better to correlate this later in some data tool, rather than adding weather data into the customer entity? – but these concerns were swept aside ("Don’t bring that negative vibe, we can do this together!") and who are we to question the business? We better get coding:
We add the temperature to our Customer class:
@JsonInclude(JsonInclude.Include.NON_NULL)
public class Customer {
... snip ...
private Double airTempActual;
private Double airTempFeelsLike;
}Someone duck duck go’ed "Weather API" and we found a suitable service with the right price of $0 features for our needs.
I’m using the Openweathermap API, which has a very nice developer experience and is simple and intuitive. 5 stars, would recommend.
We knocked up a class to hold the response from the API:
@NoArgsConstructor
@Getter
public class Weather {
private WeatherMain main;
}
@NoArgsConstructor
@Getter
class WeatherMain {
private double temp;
@JsonProperty("feels_like")
private double feelsLike;
}
And wired it all up straight into createCustomer method on the CustomerController
public class CustomerController {
private final CustomerDAO customerDAO;
private final RestTemplate restTemplate;
@Value("${WEATHER_API_KEY}")
private String weatherApiKey;
... snip ...
@PostMapping
public ResponseEntity<Void> createCustomer(@RequestBody Customer customer) throws Exception {
// whats the temp in Perth?
try {
var perthWeather = perthWeather();
customer.setAirTempActual(perthWeather.getMain().getTemp());
customer.setAirTempFeelsLike(perthWeather.getMain().getFeelsLike());
} catch (Exception e) {
log.warn("cant get the weather in perth", e);
}
this.customerDAO.save(customer);
return ResponseEntity.created(new URI("/" + customer.getId().toString())).build();
}
/**
* call the weather api to get the weather in Perth
* @return
*/
private Weather perthWeather() {
try {
var query = Map.of("appid", weatherApiKey, "q", "perth", "units", "metric");
return this.restTemplate.exchange("https://api.openweathermap.org/data/2.5/weather?q={q}&appid={appid}&units={units}", HttpMethod.GET, null, Weather.class, query).getBody();
} catch (RestClientException e) {
log.error("failed to get the weather in perth: "+ e.getMessage(), e);
throw new RuntimeException(e);
}
}
Let’s push to prod! Oh, after following the CAB process and presumably some sort of testing. In fact, I’ll even test it on my machine:
It starts and we can see the temps are null on my bootstrapped data.
2020-07-06 16:35:17.243 INFO 55154 --- [ main] c.m.n.NotreactiveDbApplication : Started NotreactiveDbApplication in 5.739 seconds (JVM running for 6.643)
Customer(id=1, firstName=Brock, lastName=Mills, airTempActual=null, airTempFeelsLike=null)
Customer(id=2, firstName=Brian, lastName=Fitzgerald, airTempActual=null, airTempFeelsLike=null)That’s fine, we don’t want to fail a Customer create just because we didn’t know the weather in some random city.
But does it work if we add a new customer?
brockmills@emmet microservices % curl --header 'Content-Type: application/json' --data '{"firstName":"Tester", "lastName":"Testing"}' http://localhost:8080/
brockmills@emmet microservices % curl http://localhost:8080/
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"},{"id":3,"firstName":"Tester","lastName":"Testing","airTempActual":17.6,"airTempFeelsLike":10.3}]17 degrees in Perth. A lovely winter’s day and lovely microservice to boot (to Spring Boot, if you will hardi har har).
But this is not a blog about the weather, it’s about reactive! The architect re-appeared from wherever they go when they are not bothering engineers, and they had done some detailed usage projections. Turns out we are going to need to support 4x our expected load and – get this – we can’t deploy any new servers. We have to work with what we have got.
We took some stats from prod, created some flame graphs: 
We got confused about the output3 so tried some other profilers, which produced some actual useful output: 
We managed to produce a flame graph of the same thing: 
Boom- there’s our ticket to doing a APM presentation at DevHatMineCon next year!
We now had some pretty strong evidence that most of the thread time on our app was spent waiting for the weather API (and a little at the database as well).
We also found a neat natural experiment to reinforce our hypothesis: We accidently ran production against our mock Weather API server for 3 hours4. That thing is fast and it was still spending 90% of its time talking to the API.
We need to go nuclear – with the reactor!
This section builds on reactive framework concepts we discussed in part 2
Porting the application to reactive is relatively straight forward.
There are 3 main concerns:
- We need to replace the JPA and JDBC database components with reactive enabled ones;
- We need to wrap our Controller method return types in either
MonoorFluxcontainers; and - We need to replace the
RestTemplateclient with a reactiveWebClient.
Let’s get into it:
I’ll start with the project dependencies. spring-boot-starter-web and spring-boot-starter-data-jpa are out, to be replaced with spring-boot-starter-webflux and spring-boot-starter-data-r2dbc.
The database bound beans require small changes to remove JPA annotations, in the case of my Customer, this means removing @Entity and @GeneratedValue:
@Data
@AllArgsConstructor
@NoArgsConstructor
@JsonInclude(JsonInclude.Include.NON_NULL)
public class Customer {
@Id
private Integer id;
private String firstName;
private String lastName;
private Double airTempActual;
private Double airTempFeelsLike;
}Q: If we remove
@GeneratedValueannotation, how does the primary key get updated?
A: It depends. The reactive Spring Data library really only supports the database sequence / identity approach from JPA, which is driven by the@IDannotation. As such it is aware of database sequences on fields with the@IDand manages them.
The DAO changes from implementing a JpaRepository to a ReactiveCrudRepository.
public interface CustomerDAO extends ReactiveCrudRepository<Customer, Integer> {
}Pretty painless so far.
The Initialiser class is the first taste of the real differences. Firstly, since we are not using JPA / Hibernate, there is no auto generation of database schemas, so we are going to do this manually with the r2dbc provided DatabaseClient.
Next, to bootstrap the test data we need to create it in the context of a reactive type, being a Flux for the list of customers.
To put it all together:
@RequiredArgsConstructor
@Component
class Initialiser implements ApplicationListener<ApplicationReadyEvent> {
private final CustomerDAO customerDAO;
private final DatabaseClient databaseClient;
@Override
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
Flux<Customer> customers = Flux.just("Brock:Mills", "Brian:Fitzgerald")
.map(s -> {
var c = s.split(":");
return new Customer(null, c[0], c[1], null, null);
})
.flatMap(customerDAO::save);
databaseClient.execute("create table CUSTOMER(ID identity auto_increment, FIRST_NAME varchar(50), LAST_NAME varchar(50), AIR_TEMP_ACTUAL double, AIR_TEMP_FEELS_LIKE double)")
.fetch()
.rowsUpdated()
.thenMany(customers)
.thenMany(this.customerDAO.findAll())
.subscribe(System.out::println);
}
}This is interesting. customers is a flux that is setup, but (because we are reacting) nothing happens with it until we hit the subscribe() down the track.
This is demonstrated nicely by adding in a few log.info() prints:
@Override
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
Flux<Customer> customers = Flux.just("Brock:Mills", "Brian:Fitzgerald")
.map(s -> {
log.info("1: creating customer: " + s);
var c = s.split(":");
return new Customer(null, c[0], c[1], null, null);
})
.flatMap(customerDAO::save);
log.info("2: this is after the customer");
databaseClient.execute("create table CUSTOMER(ID identity auto_increment, FIRST_NAME varchar(50), LAST_NAME varchar(50), AIR_TEMP_ACTUAL double, AIR_TEMP_FEELS_LIKE double)")
.fetch()
.rowsUpdated()
.thenMany(customers)
.thenMany(this.customerDAO.findAll())
.subscribe(c -> {
log.info("3: in the subscribe: " + c.toString());
});
log.info("4: the end, we are initialised");
}Results look like this:
[ main] c.m.reactivedb.ReactiveDbApplication : Started ReactiveDbApplication in 5.608 seconds (JVM running for 6.9)
[ main] com.marlo.reactivedb.Initialiser : 2: this is after the customer
[ main] com.marlo.reactivedb.Initialiser : 1: creating customer: Brock:Mills
[ main] com.marlo.reactivedb.Initialiser : 1: creating customer: Brian:Fitzgerald
[ main] com.marlo.reactivedb.Initialiser : 3: in the subscribe: Customer(id=1, firstName=Brock, lastName=Mills, airTempActual=null, airTempFeelsLike=null)
[ main] com.marlo.reactivedb.Initialiser : 3: in the subscribe: Customer(id=2, firstName=Brian, lastName=Fitzgerald, airTempActual=null, airTempFeelsLike=null)
[ main] com.marlo.reactivedb.Initialiser : 4: the end, we are initialised
The customers flux is set up, but there’s no subscription so execution of the stream is deferred. The 2nd statement, at the end of the databaseClient.execute() chain, does subscribe to the stream which causes the publisher to execute, create the DB tables and insert the customer records and finally log the resulting records5.
The CustomerController is where the rubber really hits the road. Firstly we need to replace RestTemplate with the WebClient HTTP Client and modify the return types to use the reactive containers. We do retain the @RestController annotation though, as it is reactive aware:
@RestController
@RequiredArgsConstructor
@Slf4j
public class CustomerController {
private final CustomerDAO customerDAO;
private final WebClient webClient;
...snip...
@GetMapping("/")
public Flux<Customer> getCustomers() {
return this.customerDAO.findAll();
}
@PostMapping("/")
public Mono<ResponseEntity<Void>> createCustomer(@RequestBody Customer customer) throws Exception {
}The List on the getCustomers method becomes a Flux but the ResponseEntity on the createCustomer method must be wrapped in a Mono like this: Mono;.
Now we get to the createCustomer method. A quick recap, This API:
- calls the weather API,
- merges the weather data into the
Customerfrom the client, - saves the
Customerin the Database - returns a
201 CreatedHTTP code that includes aLocationheader with a link to the newly createdCustomer. - a failure for the weather API call should not be fatal.
Before I go into the controller method, let’s focus on the call to the Weather API in perthWeather(). This will now join the reactive party with a Mono wrapper, however I need to restructure the way this method fails. Whereas it previously simply threw an Exception that was handled by the createCustomer method, there’s a better way work now we need to operate within the reactive container.
perthWeather becomes this:
private Mono<Optional<Weather>> perthWeather() {
var q = Map.of("appid", weatherApiKey,
"q", "perth",
"units", "metric");
return webClient.get().uri("https://api.openweathermap.org/data/2.5/weather?q={q}&appid={appid}&units={units}", q)
.retrieve()
.bodyToMono(Weather.class)
.map(Optional::of)
.onErrorReturn(Optional.empty());
}WebClient has a friendly API, where we define:
- The method
get() - the target URI with template params along with a
Mapof the query parameters - tell it what to do with the body –
bodyToMono(Weather.class)
We could stop there and simply return the Mono however given the try/catch approach that was used before isn’t going fly, I’m now going to wrap the return Weather class in an Optional and add an error handler that will return an empty Optional if something goes wrong.
The createCustomer method now needs to be refactored. We now need to work within the reactive stream, which means using the methods provided by the Mono to do our businesses bidding.
Here’s the code:
@PostMapping("/")
public Mono<ResponseEntity<Void>> createCustomer(@RequestBody Customer customer) throws Exception {
return perthWeather().map(o -> {
o.ifPresent(w -> {
customer.setAirTempActual(w.getMain().getTemp());
customer.setAirTempFeelsLike(w.getMain().getFeelsLike());
});
return customer;
})
.map(customerDAO::save)
.map(c -> ResponseEntity.created(UriComponentsBuilder.fromPath("/" + customer.getId().toString()).build().toUri()).build());
}First up, call the perthWeather() method, then use the resulting Weather to fill out our Customer object weather fields. Here I’m calling map(), using the Optional container to determine if we can actually add the weather and then return the Customer, ready for the database. Just like that, I’ve removed a try catch from my method, which definitely looks cleaner. Then we map() again to save the record to the database and finally map to compose the response object including the location path.
Does it work?
brockmills@emmet ~ % curl http://localhost:8080
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"}]%
brockmills@emmet ~ % curl --header 'Content-Type: application/json' --data "{\"firstName\":\"Tester\", \"lastName\":\"$(cat /dev/urandom | env LC_CTYPE=C tr -dc 'a-zA-Z0-9' | fold -w 32 | head -1)\"}" http://localhost:8080/
brockmills@emmet ~ % curl http://localhost:8080
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"},{"id":3,"firstName":"Tester","lastName":"MEz8ibpIWSL34vWQM285aGmnrdidH7qL"}]%
brockmills@emmet ~ %Sort of! The new record is now in the db, but there’s no temperature…
Hmm, so it looks like the API call fails for some reason, however the exception is swallowed and overridden by the onErrorReturn(), leaving us without the all important weather data. Worse is that we might have let this slip into prod if we weren’t being so careful6.
The reactive API also has a onErrorResume() method which allows for more flexible error handling. Lets try that, replacing the onErrorReturn():
.onErrorResume(e -> {
log.error("error calling weather API: " + e.getMessage(), e);
return Mono.just(Optional.empty());
});Now we still default to the optional, however we log the exception from the web client, so we now have a fighting chance of working out what’s gone wrong.
Testing again results in:
2020-07-07 12:57:12.755 ERROR 2894 --- [ctor-http-nio-4] com.marlo.reactivedb.CustomerController : error calling weather API: 401 Unauthorized from GET https://api.openweathermap.org/data/2.5/weather?q=perth&appid=weatherkey&units=metric
org.springframework.web.reactive.function.client.WebClientResponseException$Unauthorized: 401 Unauthorized from GET https://api.openweathermap.org/data/2.5/weather?q=perth&appid=weatherkey&units=metric
at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:181) ~[spring-webflux-5.2.7.RELEASE.jar:5.2.7.RELEASE]
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
|_ checkpoint ⇢ 401 from GET https://api.openweathermap.org/data/2.5/weather?q=perth&appid=weatherkey&units=metric [DefaultWebClient]
Stack trace:
at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:181) ~[spring-webflux-5.2.7.RELEASE.jar:5.2.7.RELEASE]
at org.springframework.web.reactive.function.client.DefaultClientResponse.lambda$createException$1(DefaultClientResponse.java:206) ~[spring-webflux-5.2.7.RELEASE.jar:5.2.7.RELEASE]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) ~[reactor-core-3.3.6.RELEASE.jar:3.3.6.RELEASE]
at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92) ~[reactor-core-3.3.6.RELEASE.jar:3.3.6.RELEASE]
... cut for brevity, however all classes below are reactor.netty.channel, reactor.core.publisher, io.netty or java.lang namespaces..Ah, the API key isn’t being set and is defaulting to weatherapikey. Whoops, oh well that’s easy fixed, I need to add the WEATHER_API_KEY env variable.
An interesting effect of the reactive setup then subscribe/execute paradigm is that the stack trace in the exception is not particularly useful in locating the source of the error. It does identify the class where the thing has gone wrong, but since my code is really just defining the execution rather than getting executed, it does nothing to help me find the line in my code that’s blown up. This is annoying, however with careful logging, shouldn’t be a blocker to running reactor in production.
Right, add the API key:
brockmills@emmet ~ % export WEATHER_API_KEY=the_api_key java -jar target/reactive-db-0.0.1-SNAPSHOT.jar
... snip ...
2020-07-07 13:02:35.163 INFO 91619 --- [ main] c.m.reactivedb.ReactiveDbApplication : Started ReactiveDbApplication in 4.838 seconds (JVM running for 11.345)and retest:
brockmills@emmet ~ % curl --header 'Content-Type: application/json' --data "{\"firstName\":\"Tester\", \"lastName\":\"$(cat /dev/urandom | env LC_CTYPE=C tr -dc 'a-zA-Z0-9' | fold -w 32 | head -1)\"}" http://localhost:8080/
brockmills@emmet ~ % curl http://localhost:8080
[{"id":1,"firstName":"Brock","lastName":"Mills"},{"id":2,"firstName":"Brian","lastName":"Fitzgerald"},{"id":3,"firstName":"Tester","lastName":"31VqMkuXXVRa06lZsfXiNTdgZYKkwjIP","airTempActual":16.72,"airTempFeelsLike":16.18}]%It works! Deploy and we are done. Get it into prod and bask in the glow of simple, performant, modern microservices.
Hopefully this has provided a reasonable introduction into how reactive does its thing and what microservices using Spring Boot Reactive look like.
CompuHyperGlobalMegaNet will surely dominate with this, but we aren’t finished. Oh no, not even close. We recently hired an intern and thought we could get them started on unit tests and error handling and one of the BA’s asked what the new performance stats were like on reactive (did we re-run the test? Might need to follow that up) so keep an eye out for posts in the near future.
The code is available on gitlab at (https://gitlab.com/themarlogrouppublic/reactive-blogs).
Endnotes
- Richer than astronauts!
- I mean start, not golang. We are enterprise, take this seriously, please.
- Lots of flame graph tools measure CPU usage and computers are smart enough to not worry the CPU with IO tasks. I was actually surprised at the amount of time the JVM spends in the compiler here and wonder how long I would need to run it for to get the majority of samples to be business logic rather than JVM. I sense a follow up post!
- Yes, it did (but with something far more important than this. In my defense the client was using my test environment 🙃).
- Calls to subscribe() are meant to be non blocking, however in the log here we can see these are all executing on the main thread, i.e. the thread is blocked. I think this is a spring context specific thing, but I don’t actually know why this happens. ¯\(ツ)/¯
- We wont have to be nearly as careful once we get those automated tests the QA team have been banging on about for ages.