DEV Community: ndr_brt

Zero Downtime Deployment with Docker Swarm

ndr_brt — Mon, 21 Dec 2020 12:54:27 +0000

If you are a software developer that in the past has dealt with production software, you are certainly familiar with this struggle:

deployment time

I hope you have already put into effect measures to trust your deployment by taking advantage of great practices such as Continuous Integration, Automated Testing, Continuous Delivery and so on, these should be carried out before tackling the thing I’m going to talk about. This is not going to be a silver bullet.

Today we’ll talk about zero downtime.

the problem

Imagine you have already packaged your application into a docker image, published it into a docker registry and have it up and running in production by the following command:

docker run …

Nothing wrong, your application is good to go.

But… how can we update it?

The easiest solution is to stop the old one and start the new version

docker stop <running container id>

docker run <new version>

There’s a problem with this approach: from the time you stop the old container and the complete bootstrap of the new version, your application will not respond.

the solution

This problem is pretty common and can be solved in many ways. We discovered an almost “effort free” way using Docker Swarm.

what is Docker Swarm?

Swarm is a Docker “mode” already included in your Docker installation. It’s a powerful cluster engine that will help you scale your application.

Also, it will solve your downtime problems.

how?

The idea is to transform your docker instance into a single node swarm cluster:

docker swarm init

This command should return something like

Swarm initialized: current node (<node_id>) is now a manager.

To add a worker to this swarm, run the following command:

docker swarm join — token <swarm_token> <node_addess+port>

To add a manager to this swarm, run ‘docker swarm join-token manager’ and follow the instructions.

These are instructions for adding nodes to our cluster. But you don’t need that now.

Now you need to deploy the stack.

In order to achieve that, you need to define a docker-compose file like this:

version: ‘3.7’

networks:
  my-network:
    external: false

services:
  my-server:
    image: ${IMAGE}
    hostname: my-server
    container_name: my-server
    ports:
    - ‘8080:8080’
    networks:
    - my-network
    healthcheck:
      test: [“CMD”, “curl”, “-i”, "http://localhost:8080/health"]
    deploy:
      mode: replicated
      replicas: 2
      update_config:
        order: start-first
        failure_action: rollback
        delay: 5s

In this example there is just one container, but you can deploy as many as you need.

Some concepts:

— image: a variable that represents the image name and version

— network: the stack network, with this various services can communicate with each other

— healthcheck: definitions of commands that verify the service status (up or down). The commands should return error code = 0 when the service status is good and error code != 0 when the service has not started yet, stopped, paused, etc..

— replicas: the numbers of parallel containers of the service that will be deployed

You can start your service by the following command:

export IMAGE=${IMAGE_NAME}:${IMAGE_VERSION}

docker stack deploy -c docker-compose.yml <stack_name> — with-registry-auth

Once you are ready to update your service, you can use the same command, with a different value of IMAGE_VERSION, and there will be no downtime since Swarm will take care of starting the new containers and, once started correctly, it will also take care of stopping the old ones.

Throttle HTTP requests on paged resources with Vert.x

ndr_brt — Mon, 21 Dec 2020 12:53:40 +0000

Hey Java developers, I know that you are interested in asynchronous programming and I know you are struggling with such paradigm.
I know that because I’m one of you!

Let us see how to deal with rate limiting issues in the magical async world in Java with Vert.x.

For the unaware a little recap:

What is Vert.x?

Vert.x is a polyglot event-driven and non-blocking application framework.

What is “Rate limiting”?

Rate limiting is a set of policies implemented by a web server used to force client to “throttle” requests from being overloading.

What are paged resources?

Paged resources are resources split in various pages to prevent overloading.

We will work with a real implementation case:

We are music fanatics (as I am), we are users of Discogs.com, and we want to fetch its data in our non-blocking application.

Taking a look at the discogs.com API reference we will notice that — aside from the API specifications — there are some information about data format, headers, versioning and so on.
We are now interested in two sections: Rate Limiting and Pagination.
Those are the problems to deal with.

A sketch of our architecture

Every arrow from left to right is a method call, every arrow from right to left is a Future.

We will start from the bottom and go up the river.

Rate Limiting

Discogs permits, according with documentation, 25 request per minute, so, 25 will be our rate limit, and one minute is the size of the rate limit window.

The idea is to have an interface with a method called *execute *which will accept a Request (a data object that represents the request) as parameter and return a Future (because we think async!) of Buffer, that is the response’s body.
Trivial and neat.

interface Requests {
  Future<Buffer> **execute**(Request request);
}

We need to put a delay between requests, hence inside the RequestExecutor implementation we will use a queue to decouple the execute invocations from the throttled request execution:

class ThrottledRequests implements Requests {

  ...

  @Override
  public Future<Buffer> execute(Request request) {
    try {
      Future<Buffer> future = Future.future();
      queue.put(new RequestContext(request, future));
      return future;        
    } catch (Exception e) {
      return Future.failedFuture(e);
    }
  }

  ...

}

RequestContext is a simple data object that allow us to bind request to its future.

Then, the request should be consumed from the queue and executed, we will do this with a periodic task by starting it every N milliseconds, where N is a non-fixed value computed from the maximum number of requests we are allowed to do in the rate limit window.

At first we will start with an optimistic 1, therefore here’s our ThrottledRequests constructor:

public class ThrottledRequests implements Requests {
  private static final int RATE_LIMIT_WINDOW_SIZE = 60000;
  private final Logger log = getLogger(getClass());
  private final BlockingQueue<RequestContext> queue = new LinkedBlockingQueue<>();
  private final AtomicLong executorId = new AtomicLong(0);
  private final AtomicLong actualDelay = new AtomicLong(100);
  private final HttpClient http;
  private final Vertx vertx;

  ThrottledRequests(Vertx vertx) {
    this.vertx = vertx;
    this.http = vertx.createHttpClient();
    long id = vertx.setPeriodic(actualDelay.get(), executor());
    this.executorId.set(id);
  }

  ...
}

Some explanations:

vertx: you need the vertx instance to launch a periodic executor.
actualDelay: is an AtomicLong that represent the actual delay, so, as we said before, it’s starting value is 1.
executor(): core method of the class, it insantiate a function that will perform the throttled requests.
executorId: we need to track the executor task id to stop it when the actualDelay will change. Is an AtomicLong too

Let us see how to handle async request executions:

private Handler<Long> executor() {
  return timerId -> {
    RequestContext context = queue.poll();
    if (context != null) {
      Request inputRequest = context.request;
      http.request(inputRequest.method(), inputRequest.options())
        .putHeader("User-Agent", "YourApp/0.1")
        .setFollowRedirects(true)
        .handler(response -> {
          response.bodyHandler(context.future::complete);
          checkAndUpdateRateLimit(response);
        })
        .end();
    }
  };
}

Easy, uh?
On every request responded we complete the future and…checkAndUpdateRateLimit():
according to the API documentation, there is a header in every response we will get from discogs called X-Discogs-Ratelimit, that tells us how many requests we can do in a rate limit window of 1 minute. Cool.

private void checkAndUpdateRateLimit(HttpClientResponse response) {
  Optional.ofNullable(response.getHeader("X-Discogs-Ratelimit"))
    .map(Long::parseLong)
    .map(rateLimit -> rateLimit - 1)
    .map(reqPerMinute -> RATE_LIMIT_WINDOW_SIZE / reqPerMinute)
    .ifPresent(throttleDelay -> {
      if (throttleDelay != actualDelay.getAndSet(throttleDelay)) {
        vertx.cancelTimer(executorId.get());
        long id = vertx.setPeriodic(throttleDelay, executor());
        executorId.set(id);
      }
    });
}

The calculation is pretty simple:

Check the rate limit header
Subtract one, just to avoid to fill the window
Calculate the delay you should take between requests to do less than rateLimit requests in the time window (one minute)
If the delay is different from the current one, we remove the old executor’s timer and define another one with a new delay time.

Pagination

Handling pagination is a little trickier than rate limit, since there is some work to do with futures, you know, the total page count will be known only after fetching the first page.

The idea is to divide responsibilities, we will implement three components:
(the example is on the Inventory resource, which retrieves Listing entities)

GetListingPage: one implementation for every resource, such class uses Requests as collaborator, knows the url to invoke and knows how to deserialize the single fetched page. So its responsability is the Page.
Pages uses GetResourcePage to fetch first page and eventually the others, join them. Its responsibility is the Pages.
Client obtains all the pages from Pages and extract the entities from those.

Let’s start from GetListingPage, is a class that implement an interface with one method: apply

@Override
Future<ListingPage> apply(String userId, Integer pageNumber) {
  log.info("Request {} inventory page {}", userId, pageNumber);
  String path = String.format("/users/%s/inventory?page=%d", userId, pageNumber);
  Request request = Request.get(path);

  Future<ListingPage> future = Future.future();
  executor.execute(request).setHandler(async -> {
    if (async.succeeded()) {
      String json = async.result().toString();
      ListingPage page = Json.fromJson(json, ListingPage.class);
      future.complete(page);
    } else {
      future.fail(async.cause());
    }
  });

  return future;
}

Nothing special, execute request and deserialize JSON output to ListingPage class (a class that represent the ListingPage structure).

Maybe the more interesting class is Pages. The public method is getFor:

public Future<List<T>> getFor(String userId) {
  Future<List<T>> result = Future.*future*();

  getPage.apply(userId, 1).setHandler(async -> {
    if (async.succeeded()) {
      T firstPage = async.result();
      int totalPages = firstPage.pages();
      log.info("Total pages: {}", totalPages);

      List<Future> futures = IntStream
        .range(2, totalPages + 1)
        .mapToObj(page -> getPage.apply(userId, page))
        .collect(Collectors.toList());

      CompositeFuture.all(futures)
        .setHandler(joinPages(result,firstPage));
    }
  });

  return result;
}

We will ask for the first page and then, knowing the total pages count, we’ll ask for every other page, combining all the Futures into a CompositeFuture.

When every future will be completed, we will joinPages:

private Handler<AsyncResult<CompositeFuture>> joinPages(
  Future<List<T>> future, T firstPage
) {
  return async -> {
    if (async.succeeded()) {
      CompositeFuture remnants = async.result();
      Collection remnantPages = IntStream
        .range(0, remnants.size())
        .mapToObj(remnants::resultAt)
        .map(Page.class::cast)
        .collect(Collectors.toList());

      List<T> total = new ArrayList<>();
      total.add(firstPage);
      total.addAll(remnantPages);

      future.complete(total);
    } else {
      future.fail(async.cause());
    }
  };
}

Now, at the higher level (Client) we can just get pages and extract entities:

listingPages.getFor(user).setHandler(async -> {
  if (async.succeeded()) {
    List<Listing> listings = async.result().stream()
      .map(ListingPage::listings)
      .flatMap(Collection::stream)
      .collect(Collectors.*toList*()));

    ... do your logic ...

  }
});

Pretty cool, huh?

Source code:

TDD in an event driven application

ndr_brt — Mon, 21 Dec 2020 12:52:53 +0000

You (should) know, Test Driven Development is one of the greatest practice in software development: it gives you confidence in what you implement, it gives you courage to refactor or change behaviors, it forces you to think about what you are about to implement.

For sure not every piece of code deserves to be written test-first, but at least you should start thinking: “Can I write a test before?”.

Passing to an event driven world the process remains the same:
there’s a SUT, a trigger action, and some behavior verification:

Consider implementing a feature for an hypothetical Continuous Integration application, on tests executed should check the results and notify to the application. This application in based on Vert.x framework and is coded in Kotlin language.

So our first test will be: “on tests executed, when they passed, should emit tests passed”:

@ExtendWith(VertxExtension::class)
internal class CheckTestsExecutionTest {

  @Test
  @Timeout(3, timeUnit = SECONDS)
  internal fun `on tests executed, when they passed, should emit tests passed`(vertx: Vertx, context: VertxTestContext) {

    CheckTestsExecution(vertx.eventBus()).start()

    vertx.eventBus().consumer<JsonObject>("TestsPassed") {
      context.completeNow()
    }

    vertx.eventBus()
      .publish("TestsExecuted", JsonObject.mapFrom(
        TestsExecuted(listOf(
          TestResult("simple test", passed = true)
        )))
      )
  }
}

There’s a lot of stuff in there. The VertxExtension *class is a junit5 extension provided by the library *vertx-junit5 that gives some facilities to write Vert.x unit tests.

But, the very important thing here is that there’s no assertion. The reason is simple: we are testing an asynchronous behavior, so we will get no synchronous output from our .

What we need here is to set an expectation. To do that we create a consumer on the event we expecting to be launched from our SUT. To make test pass then we call the completeNow method on context object (is a Vert.x junit5 extension utility to handle asynchronous tests, as well as @Timeout annotation).
The test will pass only when completeNow method is called before the timeout. Simple.

The last test row emits an event that should trigger our desired behavior.
Obviously, after fixing compilation errors, the test will fail. Red bar, cool.

Let’s write only enough code to make test pass:

class CheckTestsExecution(private val eventBus: EventBus) {

  fun start() {
    eventBus.consumer<JsonObject>("TestsExecuted") {
      eventBus.publish("TestsPassed", JsonObject())
    }
  } 
}

Green bar. Go on.

Now, we need to describe the case where one test fails and so we will expect to receive TestsFailed event. We should also expect not to receive TestsPasses.

  @Test
  @Timeout(3, timeUnit = SECONDS)
  internal fun `on tests executed, when they failed, should emit tests passed`(vertx: Vertx, context: VertxTestContext) {
    CheckTestsExecution(vertx.eventBus()).start()

    vertx.eventBus().consumer<JsonObject>("TestsPassed") {
      context.failNow(RuntimeException("No TestPassed event should be emitted"))
    }

    vertx.eventBus().consumer<JsonObject>("TestsFailed") {
      context.completeNow()
    }

    val testResults = listOf(
      TestResult("simple test", passed = false)
    )
    vertx.eventBus()
      .publish("TestsExecuted", JsonObject.mapFrom(TestsExecuted(testResults)))
  }

Red bar again, so, make it green ASAP!

class CheckTestsExecution(private val eventBus: EventBus) {

  fun start() {
    eventBus.consumer<JsonObject>("TestsExecuted") { message ->
      val testExecuted = message.body().mapTo(TestsExecuted::class.java)
      if (testExecuted.tests.none { !it.passed }) {
        eventBus.publish("TestsPassed", JsonObject())
      } else {
        eventBus.publish("TestsFailed", JsonObject())
      }
    }
  }

}

Perfect, you just implemented an event driven component with TDD on Vert.x in Kotlin. Refactoring is up to you.

As you see, no great difference from the classic TDD approach: think, write a test, do something and expect behaviors, just pay attention to asynchronous manners and use expectations.

Handle backpressure between Kafka and a database with Vert.x

ndr_brt — Thu, 06 Aug 2020 06:52:39 +0000

As we already discussed in the past, asynchronous programming brings many pros in developing reactive and responsive applications. However, it also carries cons and challenges, one of the main ones is the backpressure problem.

What is backpressure?

In physics

it is a resistance or force opposing the desired flow of fluid through pipes (wikipedia)

We can translate the problem on a known scenario: persistence of messages polled from a bus, where there are a huge amount of messages on a bus that our application is polling really fast, but the database underneath is really slow.

How can be funnel overflow be avoided?

In a synchronous scenario, there's no backpressure's issue, the sync nature of the computation blocks the polling from the bus until the current message is processed.
But, in the async world, the polling is executed without clues on what's happening on the database. So if the database can't handle all the messages coming from the bus, the messages will remain "in between", which means in the memory of our service.
This can lead to failures or, at worst, to service fault.

Let's try to develop an application that persists messages in a database, and make it evolve to handle backpressure

Automatic polling

At first, our verticle will do those operations:

initialize the JDBC client
initialize the Kafka client
subscribe to the topic
persist the records

The code is quite simple, and it works well with small amounts of messages.
When the load gets bigger and bigger a problem appears: using the handler of Vertx Kafka consumer means that there's no control on the message ratio, so it will poll continuously without considering the persistance rate, causing memory overload.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise)

      .handler(record -> {

        persist(jdbc, record)

          .onSuccess(result -> System.out.println("Message persisted"))

          .onFailure(cause -> System.err.println("Message not persisted " + cause));

      });

  }

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

private Future<UpdateResult> persist(JDBCClient jdbc, KafkaConsumerRecord<String, String> record) {

    Promise<UpdateResult> promise = Promise.promise();

    JsonArray params = toParams(record);

    jdbc.updateWithParams("insert or update query to persist record", params, promise);

    return promise.future();

  }

private JsonObject datasourceConfiguration() {

    // TODO datasource configuration

    return null;

  }

private JsonArray toParams(KafkaConsumerRecord<String, String> record) {

    // TODO: convert the record into params for the sql command

    return null;

  }

}

Explicit polling

To handle the backpressure, explicit polling shall be used, and this can be done by avoiding the kafka consumer's handler setting and by calling poll manually (in the following case, every 100ms).
By using this approach, it can be made so that every poll gets performed only when the batch of previously polled messages are persisted.
This behaviour can be achieved by handling every message's persist future and collecting all of them with the CompositeFuture.all, that will succeed only when all the messages are completed, and only in this case the next polling can be made.
If at least one of the future fails, everything will fail, and the polling will stop.
There are various solutions that can be adopted to make the service handle the failure, e.g. sending the message to a Dead Letter Queue, but we will not cover this case.

The problem with this code is that if a message fails, we will lose it, because the consumer is set to auto-commit, so, it's vertx that commits the topic offset.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer<String, String> consumer = KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise);

<span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>



}

private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">List</span><span class="o">&lt;</span><span class="nc">Future</span><span class="o">&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;&gt;</span> <span class="n">futures</span> <span class="o">=</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
      <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
      <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="n">record</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">))</span>
      <span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">toList</span><span class="o">());</span>

    <span class="k">return</span> <span class="nc">CompositeFuture</span><span class="o">.</span><span class="na">all</span><span class="o">(</span><span class="k">new</span> <span class="nc">ArrayList</span><span class="o">&lt;&gt;(</span><span class="n">futures</span><span class="o">));</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">))</span>
<span class="o">;</span>



}

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

...

}

Manual commit

Setting the ENABLE_AUTO_COMMIT_CONFIG properties to false, the service takes ownership of the topic offset commit.
The commit will be performed only when every message will be persisted, with this trick the at least once delivery is achieved.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer<String, String> consumer = KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise);

<span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>



}

private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">List</span><span class="o">&lt;</span><span class="nc">Future</span><span class="o">&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;&gt;</span> <span class="n">futures</span> <span class="o">=</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
      <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
      <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="n">record</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">))</span>
      <span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">toList</span><span class="o">());</span>

    <span class="k">return</span> <span class="nc">CompositeFuture</span><span class="o">.</span><span class="na">all</span><span class="o">(</span><span class="k">new</span> <span class="nc">ArrayList</span><span class="o">&lt;&gt;(</span><span class="n">futures</span><span class="o">));</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">Promise</span><span class="o">&lt;</span><span class="nc">Void</span><span class="o">&gt;</span> <span class="n">commitPromise</span> <span class="o">=</span> <span class="nc">Promise</span><span class="o">.</span><span class="na">promise</span><span class="o">();</span>
    <span class="n">consumer</span><span class="o">.</span><span class="na">commit</span><span class="o">(</span><span class="n">commitPromise</span><span class="o">);</span>
    <span class="k">return</span> <span class="n">commitPromise</span><span class="o">.</span><span class="na">future</span><span class="o">();</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">any</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted and committed"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting and committing messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">))</span>
<span class="o">;</span>



}

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(ENABLE_AUTO_COMMIT_CONFIG, "false");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

  ...

}

Bonus feature: achieve ordering

With a little effort it's possible to achieve ordering:
future composition allows forcing every persist operation to wait the completion of its precedent.
It's achievable by chaining the async computations one to another, so every one will be executed when the precedent future succeeds.
This is a smart pattern to be used when serialization is needed.



public class MainVerticle extends AbstractVerticle {

  ...

  private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
    <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
    <span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="nc">Future</span><span class="o">.&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;</span><span class="n">succeededFuture</span><span class="o">(),</span>
      <span class="o">(</span><span class="n">acc</span><span class="o">,</span> <span class="n">record</span><span class="o">)</span> <span class="o">-&gt;</span> <span class="n">acc</span><span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">it</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">)),</span>
      <span class="o">(</span><span class="n">a</span><span class="o">,</span><span class="n">b</span><span class="o">)</span> <span class="o">-&gt;</span> <span class="n">a</span>
    <span class="o">)</span>
  <span class="o">)</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">Promise</span><span class="o">&lt;</span><span class="nc">Void</span><span class="o">&gt;</span> <span class="n">commitPromise</span> <span class="o">=</span> <span class="nc">Promise</span><span class="o">.</span><span class="na">promise</span><span class="o">();</span>
    <span class="n">consumer</span><span class="o">.</span><span class="na">commit</span><span class="o">(</span><span class="n">commitPromise</span><span class="o">);</span>
    <span class="k">return</span> <span class="n">commitPromise</span><span class="o">.</span><span class="na">future</span><span class="o">();</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">any</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted and committed"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting and committing messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">));</span>



}

  ...

}

Conclusion

Backpressure is a fundamental topic to cover when working with async programming.
It does not come for free out of the vert.x box, but it can be acheived with some simple tricks.

Future Composition in Vert.x

ndr_brt — Wed, 20 May 2020 11:42:45 +0000

For those who don't know, Vert.x is an event driven and non blocking application toolkit.
It's polyglot, so you can use it with different languages (as Java, Kotlin, JavaScript, Groovy, Ruby or Scala).

What does "non blocking" mean?

In synchronous programming, when a function is called, the caller has to wait until the result is returned.
This behaviour could lead to performance issues.

Often the "obvious solution" seems to be concurrent programming, but dealing with shared resources and threads is not easy and deadlocks are around the corner.

Explaining how Vert.x guarantees asynchrony through the event loop concept is not in the scope of this article, but anything you may want to know gets unraveled in the great Gentle guide to async programming with Vert.x.

In a "non blocking" world, when the result of a function can be provided immediately, it will be. Otherwise a handler is provided to handle it when it will be ready.

asyncFunctionCall(result -> {
    doSometingWith(result);
})

Usualy a big "but" gets raised the first time an async piece of code is seen: But... I need that result now!

This kind of programming requires a mindset change: it's necessary to know and understand the main patterns.

Futures vs Callbacks

There are two ways to deal with async calls in Vert.x.

Pass a callback that will be executed on call completion.
Handle a future returned from the function/method call.

Callback

asyncComputation(asyncResult -> {
  if (asyncResult.succeeded()) {
    asyncResult.result()
    // do stuff with result
  } else {
    // handle failure
  }
})

A callback is a function passed to an async method used to handle the result of its computation.
It's simple to implement but it brings some drawbacks:

Unit testing becomes not-so-immediate
Leads to the dreaded Callback Hell
There's more code to be written.

Future

Future future = asyncComputation()

future.onSuccess(result -> {
  // do stuff
})

future.onFailure(cause -> {
  // handle failure
})

To avoid the problems listed for the "callback way", Vert.x implements a concept called Future.

A future is an object that represents the result of an action that may, or may not, have occurred yet (cf. apidoc).

How to switch from a callback-like call to a future one

Consider the callback example shown above.
We want a Future object to take advantage of the patterns described below, but asyncComputation is a function from another library, so we cannot modify it.

We can use a Promise.
According to the apidoc, it represents the writable side of an action that may, or may not, have occurred yet.
It perfectly fits our needs:

Promise promise = Promise.promise();
asyncComputation(asyncResult -> {
  if (asyncResult.succeeded()) {
    promise.complete(asyncResult.result());
  } else {
    promise.fail(asyncResult.cause());
  }
})
return promise.future()

That's it. We transformed a callback into a future.
The Promise API's give us a way to make this code more readable:

Promise promise = Promise.promise();
asyncComputation(promise::handle)
return promise.future()

The handle method takes care of the completion or failure of the promise, given an async result.

Future patterns

The Future object implements some interesting patterns that smartly help resolving async issues:

Map
Compose
Recover

Future Mapping

For those of you who know about the map function, part of the java Stream API, this feature should be immediate to understand.

The Future's map function accepts a function that transforms the result from one type to another.

asyncComputation() // returns a Future<String>
  .map(Integer::valueOf) // returns a Future<Integer>
  .onSuccess(...)
  .onFailure(...)

Future Composition

The compose method is similar to map, but is used when the mapping is an async operation itself:

asyncComputation()
  .map(Integer::valueOf)
  .compose(id -> retrieveDataById(id)) // retrieveDataById returns a Future
  .onSuccess(...)
  .onFailure(...)

Future Recovery

Futures can succeed, but can also fail.
To handle a failed future and consider a different behaviour, recover function can be used:

asyncComputation()
  .map(Integer::valueOf)
  .recover(cause -> Future.succeededFuture(0)) // when Integer::valueOf fails, the future could be recovered with a default value
  .compose(id -> retrieveDataById(id))
  .onSuccess(...)
  .onFailure(...)

Concurrent composition

To handle multiple future results at the same time, CompositeFuture is the class needed, it provides two static factory methods:

all returns a future that succeeds if all the futures passed as parameters succeed, and fails if at least one fails.

CompositeFuture.all(futureOne, futureTwo)
  .onSuccess(compositeResult ->
    // all futures succeeded
  )
  .onFailure(cause -> 
    // at least one failed
  );

any returns a future that succeeds if any one of the futures passed as parameters succeed, and fails if all fail.

CompositeFuture.any(futureOne, futureTwo)
  .onSuccess(compositeResult ->
    // at least one succeed
  )
  .onFailure(cause -> 
    // all failed
  );

Conclusions

Future composition API in Vert.x represents a solid way to write simple and affordable async code.
Always remember:

never throw exceptions into async code, use failed future instead to handle failure behaviours.
at the end of a future composition, don't forget to handle future's successes (onSuccess) and failures (onFailure)