Forem: Madhur Ahuja

Parallelism and ordering

Madhur Ahuja — Mon, 08 Jul 2019 10:29:39 +0000

One of the disadvantages of microservice event based architectures is that there is lot of parallel processing of a single entity across multiple modules.
There are many cases we want the parallelism but at the same time, want the messages to be processed in ordered manner.

There are several design pattern to solve this problem:

Kafka solves this through partitions. A topic can be split into partitions. Messages are gauranteed to be in order within a single partition. You can have multiple consumers consuming from each partition separately providing for scalability and ordering. This design assumes that each consumer is single threaded and processing the messages one after other. In practical, scenarios this is usually not the case.
One of the solution is to use SEDA queue. Camel has a dedicated page on Parallelism and ordering
There is whitepaper titles A Scalable Architecture for Ordered Parallelism
I havn't gone thorugh it yet.

Are there other methods to handle the ordering while processing the messages in parallel? Let me know in the comments...

Setting up Java project for monitoring

Madhur Ahuja — Wed, 03 Jul 2019 06:20:11 +0000

In any Java application deployed to monitoring, it is important to have proper monitoring / alerting infrastructure setup.

Apart from the infrastructure monitoring, it is essential to setup JVM metrics as well as application monitoring.

When I say application monitoring, it means monitoring the business transactions such as no. of orders / transactions, successful orders, failed orders , user signups, email triggered etc.

For this, we have to emit events from our application. These are usually called business metrics.

Dropwizard metrics is one of the standard libraries out there which allows you to emit events.

The events can be sent to various data stores, most popular being Graphite and Influxdb

The most popular types of metrics are:

Meters - A meter measures the rate of events over time. Also called as requests per second or TPS(transactions per second). Meters also track 1-, 5- and 15- minute moving averages, also called m1_rate , m5_rate and m15_rate. Meters have mark method to indicate the event.
Counters - A counter is used to keep track of counts. For example pending jobs and total requests. Total requests etc is usually a useless parameter but tools like graphite / influxdb can provide derivate function over the counters which can give us the additional functionality of meters using counters.
Timer - A timer measures both the rate a particular piece of code is called and as well as the distribution of its duration. Usually if you are using timer, you don't need meter or counter, since those functionalities are implemented by timer as well. A good practice is to have timer for your normal execution and counter/meter for exception scenarios, so that you can visualize no. of errors or rate of errors. A timer also provides, percentile times using variables p50, p75, p95, p98, p99 and standard deviation.

When using Spring framework, Spring metrics provides tight integration with Dropwizard metrics

Its important to note that metrics-spring is a different library from Spring metrics , which is part of Spring framework itself.

The below code demonstrates a simple way to bootstrap spring application for monitoring using Dropwizard metrics and metrics-spring

buildscript {
    ext {
    }
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath 'org.springframework.boot:spring-boot-gradle-plugin:1.5.16.RELEASE'
    }
}

group 'metrics-example'
version '1.0-SNAPSHOT'

apply plugin: 'java'
apply plugin: 'org.springframework.boot'

sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    compile("org.springframework.boot:spring-boot-starter-web")
    compile 'org.projectlombok:lombok:1.16.12'
    compile group: 'io.dropwizard.metrics', name: 'metrics-core', version: '4.1.0'
    compile group: 'io.dropwizard.metrics', name: 'metrics-graphite', version: '4.1.0'
    compile group: 'com.ryantenney.metrics', name: 'metrics-spring', version: '3.1.3'
    compile group: 'com.codahale.metrics', name: 'metrics-jvm', version: '3.0.2'
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

package app;

import com.codahale.metrics.ConsoleReporter;
import com.codahale.metrics.Counter;
import com.codahale.metrics.JmxReporter;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricFilter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.graphite.Graphite;
import com.codahale.metrics.graphite.GraphiteReporter;
import com.codahale.metrics.jvm.GarbageCollectorMetricSet;
import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
import com.codahale.metrics.jvm.ThreadStatesGaugeSet;
import com.ryantenney.metrics.spring.config.annotation.MetricsConfigurerAdapter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

@Component
public abstract class MetricsConfig extends MetricsConfigurerAdapter{

    private final MetricRegistry metrics = new MetricRegistry();
    Meter publishMeter = metrics.meter("publish.meter");
    Counter publishCounter = metrics.counter("publish.counter");

    @Value("${graphite.host}")
    private String graphiteHost;

    @Value("${graphite.port}")
    private int graphitePort;

    @Value("${graphite.amount.of.time.between.polls}")
    private long graphiteAmountOfTimeBetweenPolls;

    private String graphitePrefix;

    @PostConstruct
    public void run() {
        ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics)
                .convertRatesTo(TimeUnit.SECONDS)
                .convertDurationsTo(TimeUnit.MILLISECONDS)
                .build();
        reporter.start(1, TimeUnit.SECONDS);

    }

    abstract protected void configureReporters();

    protected void configureReporters(String graphitePrefix) {
        this.graphitePrefix = graphitePrefix;
        configureReporters(metrics);
    }


    @Override
    public void configureReporters(MetricRegistry metricRegistry) {
        registerReporter(JmxReporter.forRegistry(metricRegistry)
                .build()).start();
        GraphiteReporter graphiteReporter =
                getGraphiteReporterBuilder(metricRegistry)
                        .build(getGraphite());
        registerReporter(graphiteReporter);
        graphiteReporter.start(graphiteAmountOfTimeBetweenPolls,
                TimeUnit.MILLISECONDS);
    }

    private GraphiteReporter.Builder getGraphiteReporterBuilder(MetricRegistry
                                                       metricRegistry) {
        metricRegistry.register("gc", new GarbageCollectorMetricSet());
        metricRegistry.register("memory", new MemoryUsageGaugeSet());
        metricRegistry.register("threads", new ThreadStatesGaugeSet());
        return GraphiteReporter.forRegistry(metricRegistry)
                .convertRatesTo(TimeUnit.SECONDS)
                .convertDurationsTo(TimeUnit.MILLISECONDS)
                .filter(MetricFilter.ALL)
                .prefixedWith(graphitePrefix);
    }

    private Graphite getGraphite() {
        return new Graphite(new InetSocketAddress(graphiteHost,
                graphitePort));
    }
}

package app;

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class ProdMetricsConfig extends MetricsConfig {

    private static final String GRAPHITE_PREFIX =
            "collectd/graphite-monitoring-example/production";


    @Override
    protected void configureReporters() {
        configureReporters(GRAPHITE_PREFIX);
    }

    @PostConstruct()
    public void init() {
        configureReporters();
    }
}

Here is an example of dashboard visualized in Grafana

The above complete project is available on my github repository

The repo contains a dockerized spring boot application, grafana and a graphite instance.

Assuming you have docker installed, just execute below commands to get the project up and running.

./gradlew build
docker build -t metrics-example .
docker-compose up -d

Use http://localhost:3000 to browse the grafana portal and setup your dashboards.

Do let me know if you have any suggestions / feedback

Consuming from Kafka

Madhur Ahuja — Sat, 22 Jun 2019 14:01:31 +0000

I have been exploring on the best ways to consume from Kafka topic in Java. There are several ways:

1 The simplest way is using KafkaListener

@Slf4j
@Component
public class ExampleConsumer {

    @KafkaListener(id = "fooGroup", topics = "Topic2")
    public void listen(String in) {
        log.info("Received: " + in);
        if (in.startsWith("foo")) {
            throw new RuntimeException("failed");
        }
    }
}

2 The second way is to use Apache Camel . Using Apache camel is useful if you have lot of filtering logic to be applied on incoming messages and also output the processed messages onto another topic or stream.

@Component
public class CamelListener extends RouteBuilder {

    @Autowired
    private KafkaConsumerProperties kafkaConsumerProperties;

    @Override
    public void configure() throws Exception {
        from(kafkaConsumerProperties.kafkaUri()).process(exchange -> {

            String payload = exchange.getIn().getBody(String.class);
            System.out.println("Camel consumer: " + payload);
        }).end();
    }
}

3 The final and my preferred way is to use Kafka Streams

@Service
@Slf4j
public class KafkaStreamConsumer {

    @Autowired
    private KafkaConsumerProperties kafkaConsumerProperties;

    @Autowired
    private KafkaOrderFeedProcessor kafkaOrderFeedProcessor;

    private String topic;

    @PostConstruct
    public void processKafkaConsumer() {
        Properties properties = kafkaConsumerProperties.getConsumerProperties();
        KafkaStreams kafkaStreams = null;
        try {
            StreamsBuilder builder = new StreamsBuilder();
            KStream<String, String> kStream = builder.stream(topic);
            kStream.process(kafkaOrderFeedProcessor, new String[0]);
            kafkaStreams = new KafkaStreams(builder.build(), properties);
            kafkaStreams.start();
            log.info("op={}, status=OK, desc={}", "KafkaConsumer", "kafka consumer stream  started successfully");
        } catch (Exception var9) {
            log.error("op={}, status=KO, desc={} and exception={}", new Object[]{"KafkaConsumer", "exception while starting kafka consumer stream", var9.getMessage()});
            if (kafkaStreams != null) {
                kafkaStreams.close();
            }
        }

    }
}

There are various advantages of using Kafka's Streams API.

Kafka's Streams API (https://kafka.apache.org/documentation/streams/) is built on top of Kafka's producer and consumer clients. It's significantly more powerful and also more expressive than the Kafka consumer client. Here are some of the features of the Kafka Streams API:

supports exactly-once processing semantics (Kafka versions 0.11+)
supports fault-tolerant stateful processing including streaming joins, aggregations, and windowing
supports event-time processing as well as processing based on processing-time and ingestion-time has first-class support for both streams and tables, which is where stream processing meets databases; in practice, most stream processing applications need both streams AND tables for implementing their respective use cases, so if a stream processing technology lacks either of the two abstractions (say, no support for tables) you are either stuck or must manually implement this functionality yourself (good luck with that...)
supports interactive queries to expose the latest processing results to other applications and services)
more expressive: it ships with (1) a functional programming style DSL with operations such as map, filter, reduce as well as (2) an imperative style Processor API for e.g. doing complex event processing (CEP), and (3) you can even combine the DSL and the Processor API.

Considerations for high throughput kafka producer

Madhur Ahuja — Sat, 22 Jun 2019 13:59:07 +0000

I have been recently working on lot of high throughput kafka producers. Our application publishes close to 3 million kafka publishes per day. (which is still low compared to what kafka can handle)

There are some of the learnings along the way in maintaining such kafka producers:

Choose the number of partitions wisely: The number of partitions determine how much consumers can scale. Number of partitions is degree of parallelism in kafka. Kafka gives a single partition's data to single thread.

Our general thumb rule is to have partitions equal to number of consumer servers. For example, if we have cluster of 20 servers consuming from kafka topic, each server will be consuming from single partition so 20 partitions.

There are many other factors to be considered as explained here

Decide a consistent key while publishing - Messages published with the same key will be published to a single partition. A partition is logic unit of ordering of messages. So if ordering of messages is important to you, you should choose a consistent key for those messages.
Use power of asynchronous - Kafka producer is by default asynchronous unless you use a blocking call explicitly. That means that kafka publish can fail and your code would have moved past the publish method already. Kafka producer provides a callback once the server has executed the publish instruction. In this callback, the user can check for failure and retry the option or send to a dead letter queue etc. Kafka producer itself retries for 3 times but I believe that is too less and not enough for data critical applications.

Below is the sample snippet of such producer

@Autowired
@Qualifier("createKafkaSslProducerOrder")
Producer kafkaSslProducer;

public void publish(String messageKey, String payload, String topic) {

    try {

        ProducerRecord record = new ProducerRecord<>(topic, messageKey, payload);

        kafkaSslProducer.send(record, (metadata, exception) -> {

            if (Optional.ofNullable(exception).isPresent()) {
                log.error("op={}, status=KO, desc={} and exception={}",
                        new Object[] { "KafkaProducer",
                                "Error posting message to kafka topic: " + topic,
                                exception.getMessage() });
                // Send for re-processing
            }

        });

    } catch (Exception ex) {
        log.error("op={}, status=KO, desc=Error posting message to SSL kafka: {}, stackTrace={} ", LOG_OP_INFO, ex.getMessage(), ex);
        // Re-throw the exception so that status can be recorded in the database.
    }
}

In case of kafka messages, it is useful to provide a complete publish timestamp and origial modify timestamp of the message (such as db record). Using these timestamps, client can determine if the incoming message is stale or a new upate.
Initially, during development, it is very useful to store the partition and offset of the consumed messages. This can be stored in the consumer data store or application logs. Using this information, the message can be directly looked up in kafka to see the original message.

Tuning Linux servers for scalability

Madhur Ahuja — Tue, 07 Feb 2017 22:00:57 +0000

Focus on performance and scalablity is one of my primary personal and professional goal when working with tech products.

The server can be any Linux based server such as CentOS or Debian derivative such as Ubuntu.

In this post, I will outline my learnings on scaling the Linux server. Scaling here implies many things such as

Being able to open many files at once. Here files can be generally applied to concept such as open ports, threads etc and not necessary physical files
Being able to handle many concurrent network connections

Knowing Linux OS and related concepts such as Iptables is a pre-requisite.

Open files

We need to keep our file limit high for any linux production server. Check the current value using ulimit -a

We can configure this limit using /etc/security/limits.conf

    * hard nofile 300000
    * soft nofile 300000

    tomcat hard nofile 300000
    tomcat soft nofile 300000

Note that we can also specify per user limit as shown above (special limits for tomcat user)

The file descriptor limit for a running process can be seen in the following file under Max open files.

    $ cat /proc/<pid>/limits

    Max open files 30000

Ephemeral Ports

Increase the number of ephemeal ports availabl to your application. The default value is 32768 - 61000.

TIME_WAIT state

TCP connections go through lot of states, last of them is TIME_WAIT state. The default TIME_WAIT timeout is for 2 minutes, Which means you'll run out of available ports if you receive more than about 400 requests a second, or if we look back to how nginx does proxies, this actually translates to 200 requests per second.

These parameters can be tuned using these settings in /etc/sysctl.conf

    net.ipv4.ip_local_port_range = 18000    65535
    net.ipv4.netfilter.ip_conntrack_tcp_timeout_time_wait = 1

Connection Tracking

The next parameter we looked at was Connection Tracking. This is a side effect of using iptables. Since iptables needs to allow two-way communication between established HTTP and ssh connections, it needs to keep track of which connections are established, and it puts these into a connection tracking table. This table grows. And grows. And grows.

You can see the current size of this table using sysctl net.netfilter.nf_conntrack_count and its limit using sysctl net.nf_conntrack_max. If count crosses max, your linux system will stop accepting new TCP connections and you'll never know about this. The only indication that this has happened is a single line hidden somewhere in /var/log/syslog saying that you're out of connection tracking entries. One line, once, when it first happens.

A better indication is if count is always very close to max. You might think, “Hey, we've set max exactly right.”, but you'd be wrong.

What you need to do (or at least that's what you first think) is to increase max.

Keep in mind though, that the larger this value, the more RAM the kernel will use to keep track of these entries. RAM that could be used by your application.

We started down this path, increasing net.nf_conntrack_max, but soon we were just pushing it up every day. Connections that were getting in there were never getting out.

Maximum number of pending connections on a socket

During some of our initial load testing, we ran into a strange problem where we were unable to open more than approximately 128 concurrent connections at once.

After some investigation, we learned about the following kernel parameter.

    net.core.somaxconn

This kernel parameter is the size of the backlog of TCP connections waiting to be accepted by the application. If a connection indication arrives when the queue is full, the connection is refused. The default value for this parameters is 128 on most modern operating systems.

Bumping up this limit in /etc/sysctl.conf helped us get rid of the “connection refused issues on our Linux machines.

JVM thread count

A few hours after we allowed a significant percentage of production traffic to hit our server for the first time, we were alerted to the fact that the load balancer was unable to connect to a few of our machines. On further investigation, we saw the following all over our server logs.

    java.lang.OutOfMemoryError: unable to create new native thread

If you hit the JVM thread limit, chances are that there is a thread leak in your code that needs to be fixed. However, if you find that all your threads are actually doing useful work, is there a way to tweak the system to let you create more threads and accept more connections?

The answer, as always, is fun. It's interesting to discuss how available memory limits the number of threads that can be created on a JVM. The stack size of a thread determines the memory available for static memory allocation. Thus, the absolute theoretical maximum number of threads is a process's user address space divided by the thread stack size. However, the reality is that the JVM also uses memory for dynamic allocation on the heap. With a few quick tests with a small Java process, we could verify that as more memory is allocated for the heap, less is available for the stack. Thus, the limit on the number of threads decreases with increasing heap size.

To summarize, you can increase the thread count limit by decreasing the stack size per thread (-Xss) or by decreasing the memory allocated to the heap (-Xms, -Xmx).

Hi, I'm Madhur Ahuja

Madhur Ahuja — Tue, 07 Feb 2017 16:56:14 +0000

I am a programmer/technology guy. I like building web applications using all the standard technologies like HTML, CSS, JavaScript, JQuery, ASP.NET, and SQL Server.
I also know a good variety of programming languages too. .NET C# is my favourite, I know Java very well, and I know Javascript pretty well too.
For operating systems I have experience in both Linux and Windows, Linux being my current operating system.
I also have experience developing applications for Android.

Before entering the professional world, I spent some of the most fruitful years of my life studying Information Technology at Bharati Vidyapeeth's College of Engineering, Pune.
My interest in computers, internet and Linux dates back to 1997, when I was first introduced to a computer. That computer (Intel Pentium 120 MHz, 1.2GB HDD, 8MB RAM) was one of the most awesome things I had ever possessed.