When things may get out of control: circuit breakers in practice. Apache Zest and Akka.

In the previous post we have started the discussion about circuit breakers and why this pattern gained so much importance these days. We have learned about Netflix Hystrix, the most advanced circuit breaker implementation for JVM platform, and its typical integration scenarios. In this post we are going to continue exploring the other options available, starting from Apache Zest library.

Surprisingly, Apache Zest being certainly a gem, is not well-known and widely used. It is a framework for domain centric application development which aims to explore composite-oriented programming paradigm. Its roots go back to 2007, where it was born under another name, Qi4j (and became Apache Zest in 2015). It would require a complete book just to go through Apache Zest features and concepts, but what we are interested in is the fact that Apache Zest has simple circuit breaker implementation.

Let us use the same example to consume https://freegeoip.net/ REST(ful) web API and wrap the communication with this external service using CircuitBreaker from Apache Zest:

public class GeoIpService {
    private static final String URL = "http://freegeoip.net/";
    private final ObjectMapper mapper = new ObjectMapper();
    private final CircuitBreaker breaker = new CircuitBreaker(5, 1000 * 120);
 
    public GeoIpDetails getDetails(final String host) {
        try {
            if (breaker.isOn()) {
                GeoIpDetails details = mapper.readValue(get(host), GeoIpDetails.class);
                breaker.success();
                return details;
            } else {
                // Fallback to empty response
                return new GeoIpDetails();
            }
        } catch (final IOException ex) {
            breaker.throwable(ex);
            throw new RuntimeException("Communication with '" + URL + "' failed", ex);
        } catch (final URISyntaxException ex) {
            // Should never happen, but just trip circuit breaker immediately
            breaker.trip(); 
            throw new RuntimeException("Invalid service endpoint: " + URL, ex);
        }
    }

    private String get(final String host) throws IOException, URISyntaxException {
        return Request
            .Get(new URIBuilder(URL).setPath("/json/" + host).build())
            .connectTimeout(1000)
            .socketTimeout(3000)
            .execute()
            .returnContent()
            .asString();
    }
}

Essentially, this is as basic CircuitBreaker implementation as it could possible get. We configured it to have a threshold of 5 failures (which in our case means failing requests) and sleeping window of 2 minutes (120 * 1000 milliseconds). It becomes the responsibility of the application developer to report the successes and failures using success() and throwable(...) methods respectively, with the option to open the circuit breaker immediately using trip() method call. Please take a note that CircuitBreaker relies on Java synchronization mechanisms and is thread-safe.

Interestingly, the CircuitBreaker from Apache Zest uses a little bit different conventions: instead of operating on closed / open states, it treats them as on / off ones. Those are more familiar to most of us. And to finish up, basic JMX instrumentation is also available out of the box.

It requires a couple of lines to be added into the GeoIpService initialization (constructor f.e.) to register and expose managed beans:

public GeoIpService() throws Exception {
    final ObjectName name = new ObjectName("circuit-breakers", 
        "zest-circuit-breaker", "freegeoip.net");

    ManagementFactory
        .getPlatformMBeanServer()
        .registerMBean(new CircuitBreakerJMX(breaker, name), name);
}

Please do not hesitate to glance through official Apache Zest Circuit Breaker documentation, there are quite a few use cases you may found useful for your projects. The complete example is available on Github.

In case you are developing on JVM using Scala programming language, you are certainly a lucky one as there is native circuit breaker implementation available as part of Akka toolkit. For example, let us redesign our Geo IP service consumer as a typical Akka actor which is going to make HTTP call over to https://freegeoip.net/:

case class GeoIp(host: String)
case class GeoIpDetails(ip: String = "", countryCode: String = "", 
  countryName: String = "", latitude: Double = 0, longitude: Double = 0)

class GeoIpActor extends Actor with ActorLogging {
  import spray.json._
  import spray.json.DefaultJsonProtocol._
  
  implicit val materializer: ActorMaterializer = ActorMaterializer()
  
  import context.dispatcher
  import context.system
  
  val breaker = new CircuitBreaker(
    context.system.scheduler,
    maxFailures = 5,
    callTimeout = 15 seconds,
    resetTimeout = 2 minutes)
  
  def receive = {
    case GeoIp(host) => breaker
      .withCircuitBreaker(Http()
        .singleRequest(HttpRequest(uri = s"http://freegeoip.net/json/$host"))
        .flatMap {
          case HttpResponse(OK, _, entity, _) => Unmarshal(entity).to[GeoIpDetails]
          case _ => Future.successful(GeoIpDetails())
        }
      ) pipeTo sender()
  }
}

At this moment, the pattern undoubtedly is looking familiar to all of us. The only new option which Akka's CircuitBreaker brings on the table is overall call timeout: the execution will be considered as failed when it is not completed within this time period (certainly very handy addition to the circuit breaker capabilities). The withCircuitBreaker function takes care of managing the circuit breaker state around the wrapped block of the code.

The interactions with the GeoIpActor are no different from any other Akka actor:

implicit val system = ActorSystem("circuit-breakers")
implicit val timeout: Timeout = 5 seconds

val geoIp = system.actorOf(Props[GeoIpActor], "geo-ip-actor")
val result = (geoIp ? GeoIp("8.8.8.8")).mapTo[GeoIpDetails]
  
result andThen { 
  case Success(details) => log.info("GEO IP: {}", details)
  case Failure(ex) => log.error("Communication error", ex)
} 

By looking a bit deeper into CircuitBreaker documentation we could get some insights about its internals. There are actually three states the CircuitBreaker could be: open, closed and half-open. The presence of half-open state serves the purpose to perform just a single attempt to try out if the invocation is back to normal operations or not.

The code snippet looks perfect but one thing to keep in mind is how the Actor Model and absence of shared mutable state affects the circuit breaker state synchronization. To facilitate that, Akka's CircuitBreaker has a rich set of callback notifications (like onOpen, onHalfOpen, onClose) so the state changes could be broadcasted between actors. The complete project sources are available on Github.

For the curious readers, just a few closing notes about adoption of the circuit breaker implementations. Netflix Hystrix is the number one choice at the moment, particularly (but not only) because of superior support from Spring community. Obviously, Akka CircuitBreaker is a natural choice for Scala developers who build their applications on top of excellent Akka toolkit. Concerning Apache Zest Circuit Breaker, it could be used as-is (if you want to fully control the behavior) or be easily integrated as an useful extension into existing general-purpose clients. For example, Apache CXF allows to configure JAX-RS / JAX-WS clients with Failover feature, including the circuit breaker-based implementation: CircuitBreakerFailoverFeature.

Hope this series of posts extended a little bit your awareness about circuit breaker pattern and the state of its available implementations on the JVM platform. The repository with complete project samples is available on Github.

Stay resilient!

When things may get out of control: circuit breakers in practice. Hystrix.

It is amazing how tightly interconnected modern software systems are. Mostly every simple application has dependency on some external service or component, not to mention emerging at a great pace Internet of Things (or simply IoT) movement. It is good and not so at the same time, let us see why ...

There are many use cases when relying on other services, provided by someone externally or internally, makes a lot of sense (messaging, billing, taxes, payments, analytics, logistics, ...) but under the hood every such integration poses risks to our applications: they become dependent on availability and operationability of those services. Network latency, spikes of load, just banal software defects, each of these unknowns can bring our applications on its knees, making users and partners dissatisfied, to say it mildly.

The good news are there is a pattern we can employ to mitigate the risks: circuit breaker. Firstly explained in great details in the Release It! book by Michael T. Nygard, circuit breakers became the de-facto solution for dealing with external services. The idea is pretty simple: track the state of the external service on a given time interval to collect the knowledge about its availability. If the failure is being detected, circuit breaker opens, signalling that external service should better not be invoked for some time.

There are plenty of circuit breaker implementations available but because we are on JVM, we are going to talk about three of those: Netflix Hystrix, Akka and Apache Zest. To keep the posts considerably short, the topic of our discussion is going to be split in two parts: Netflix Hystrix followed by Akka and Apache Zest.

To show off circuit breakers in action, we are going to build a simple client around https://freegeoip.net/: public HTTP web API for software developers to search the geolocation of IP addresses. The client will return just brief geo-details about particular IP or hostname, wrapped into GeoIpDetails class:

@JsonIgnoreProperties(ignoreUnknown = true)
public final class GeoIpDetails {
    private String ip;
    @JsonProperty("country_code") private String countryCode;
    @JsonProperty("country_name") private String countryName;
    private double latitude;
    private double longitude;
}

So let us get started ...

Undoubtedly, Netflix Hystrix is the most advanced and thoroughly battle-tested circuit breaker implementation at the disposal of Java developers. It is built from the ground up to support asynchronous programming paradigm (heavily utilizing RxJava for that) and to have a very low overhead. It is more than just circuit breaker, it is full-fledged library to tolerate latency and failures in distributed systems, but we will touch upon basic Netflix Hystrix concepts only.

Netflix Hystrix has surprisingly simple design and is built on top of command pattern, with HystrixCommand in its core. Commands are identified by keys and are organized in groups. Before we are going to implement our own command, it is worth to talk about how Hystrix isolates the external service integrations.

Essentially, there are two basic strategies which Hystrix supports: offload the work somewhere else (using dedicated thread pool) or do the work in the current thread (relying on semaphores). Using dedicated thread pools, also known as the bulkhead pattern, is the right strategy to use in most use cases: the calling thread is unblocked, plus the timeout expectations could be set as well. With semaphores, the current thread are going to be busy till the work is completed, successfully or not (timeouts are claimed to be also supported since 1.4.x release branch but there are certain side effects).

Enough theory for now, let us jump into the code by creating our own Hystrix command class to access https://freegeoip.net/ using Apache HttpClient library:

public class GeoIpHystrixCommand extends HystrixCommand<String> {
    // Template: http://freegeoip.net/{format}/{host}
    private static final String URL = "http://freegeoip.net/";
    private final String host;
 
    public GeoIpHystrixCommand(final String host) {
        super(
            Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("GeoIp"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("GetDetails"))
                .andCommandPropertiesDefaults(
                    HystrixCommandProperties.Setter()
                        .withExecutionTimeoutInMilliseconds(5000)
                        .withMetricsHealthSnapshotIntervalInMilliseconds(1000)
                        .withMetricsRollingStatisticalWindowInMilliseconds(20000)
                        .withCircuitBreakerSleepWindowInMilliseconds(10000)
                    )
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("GeoIp"))
                .andThreadPoolPropertiesDefaults(
                    HystrixThreadPoolProperties.Setter()
                        .withCoreSize(4)
                        .withMaxQueueSize(10)
                )
        );
        this.host = host;
    }
 
    @Override
    protected String run() throws Exception {
        return Request
            .Get(new URIBuilder(URL).setPath("/json/" + host).build())
            .connectTimeout(1000)
            .socketTimeout(3000)
            .execute()
            .returnContent()
            .asString();
    }
}

The first thing to get from this snippet is that Hystrix commands have a myriad of different properties which are initialized in the constructor. Command group and key, set to "GeoIp" and "GetDetails" respectively, we have already mentioned. Thread pool key, set to "GeoIp", and thread pool properties (for example, core pool size and maximum queue size) allow to tune thread pool configuration, the default execution isolation strategy used by Hystrix. Please notice that multiple commands may refer to the same thread pool (shedding the load for example), but semaphores are not shared.

Other GeoIpHystrixCommand command properties, arguably most important ones, would need some explanation:

• executionTimeoutInMilliseconds sets the hard limit on overall command execution before timing out
• metricsHealthSnapshotIntervalInMilliseconds indicates how often the status of the underlying circuit breaker should be recalculated
• metricsRollingStatisticalWindowInMilliseconds defines the duration of rolling window to keep the metrics for the circuit breaker
• circuitBreakerSleepWindowInMilliseconds sets the amount of time to reject requests for opened circuit breaker before trying again

It is worth to mention that Hystrix has sensible default value for every property so you are not obliged to provide them. However, those defaults are quite aggressive (in a very good sense) so you may need to relax some. Hystrix has a terrific documentation which talks about all the properties (and their default values) in details.

Another option which Hystrix incorporates is fallback in case the command execution was not successful, timed out or circuit breaker is tripped. Although fallback is optional, it is very good idea to have one, in case of https://freegeoip.net/ we may just return an empty response.

    @Override
    protected String getFallback() {
        return "{}"; /* empty response */
    }

Great, we have our command, and now what? There are multiple ways Hystrix command could be invoked. The most straightforward one is just synchronous execution using execute() method, for example:

public class GeoIpService {
    private final ObjectMapper mapper = new ObjectMapper();
 
    public GeoIpDetails getDetails(final String host) throws IOException {
        return mapper.readValue(new GeoIpHystrixCommand(host).execute(), 
            GeoIpDetails.class);
    }
}

In case of asynchronous execution, Hystrix has a couple of options, ranging from bare Java's Future to RxJava's Observable, for example:

public Observable<GeoIpDetails> getDetailsObservable(final String host) {
    return new GeoIpHystrixCommand(host)
        .observe()
        .map(result -> {
             try {
                 return mapper.readValue(result, GeoIpDetails.class);
              } catch(final IOException ex) {
                  throw new RuntimeException(ex);
              }
        });
}

The complete sources of the project example is available on Github.

If your project is built on top of very popular Spring Framework, there is a terrific out-of-the box Hystrix support using convenient (auto)configuration and annotations. Let us take a quick look on the same command implementation using Spring Cloud Netflix project (certainly, along with Spring Boot):

@Component
public class GeoIpClient {
    @Autowired private RestTemplate restTemplate;

    @HystrixCommand(
        groupKey = "GeoIp",
        commandKey = "GetDetails",
        fallbackMethod = "getFallback",
        threadPoolKey = "GeoIp",  
        commandProperties = {
            @HystrixProperty(
                name = "execution.isolation.thread.timeoutInMilliseconds", 
                value = "5000"
            ),
            @HystrixProperty(
                name = "metrics.healthSnapshot.intervalInMilliseconds", 
                value = "1000"
            ),
            @HystrixProperty(
                name = "metrics.rollingStats.timeInMilliseconds", 
                value = "20000"
            ),
            @HystrixProperty(
                name = "circuitBreaker.sleepWindowInMilliseconds", 
                value = "10000"
            )
        },
        threadPoolProperties = {
            @HystrixProperty(name = "coreSize", value = "4"),
            @HystrixProperty(name = "maxQueueSize", value = "10")
        }
    )
    public GeoIpDetails getDetails(final String host) {
        return restTemplate.getForObject(
            UriComponentsBuilder
                .fromHttpUrl("http://freegeoip.net/{format}/{host}")
                .buildAndExpand("json", host)
                .toUri(), 
            GeoIpDetails.class);
    }
 
    public GeoIpDetails getFallback(final String host) {
        return new GeoIpDetails();
    }
}

In this case the presence of Hystrix command is really hidden so the client just dials with a plain, injectable Spring bean, annotated with @HystrixCommand and instrumented using @EnableCircuitBreaker annotation.

And last, but not least, there are quite a few additional contributions for Hystrix, available as part of the Hystrix Contrib project. The one we are going to talk about first is hystrix-servo-metrics-publisher which exposes a lot of very useful metrics over JMX. It is essentially a plugin which should be explicitly registered with Hystrix, for example here is one of the ways to do that:

HystrixPlugins
    .getInstance()
    .registerMetricsPublisher(HystrixServoMetricsPublisher.getInstance());

When our application is up and running, here is how it looks like in JVisualVM (please notice that the com.netflix.servo MBean is going to appear only after the first Hystrix command execution or instrumented method invocation so you may not see it immediately on application start):

When talking about Hystrix, it is impossible not to mention Hystrix Dashboard: terrific web UI to monitor Hystrix metrics in real time.

Thanks again to Spring Cloud Netflix, it is very easy to integrate it into your applications using just @EnableHystrixDashboard annotation and another project from Hystrix Contrib portfolio, hystrix-metrics-event-stream which exposes Hystrix metrics over event stream. The complete version of the Spring-based project example is available on Github.

Hopefully at this point you would agree that, essentially, every integration with external services (which are most of the time just a black boxes) introduces instability into our applications and may cause cascading failures and serious outages. With this regards, Netflix Hystrix could be a life saver, worth adopting.

In the next part we are going to look at another circuit breaker implementations, namely the one available as part of Akka toolkit and Apache Zest.

All projects are available under Github repository.