Kafka poll() vs heatbeat()

In older versions of Kafka, the consumer was responsible for polling the broker frequently to prove that it is still alive. If the consumer does not poll() within a specified time-limit, then the broker considers that consumer to be dead and starts re-balancing the messages to other consumers.

But in latest versions of Kafka Consumer, a dedicated background heartbeat thread is started. This heartbeat thread sends periodic heartbeats to the broker to say -"Hey, I am alive and kicking!..I am processing messages and will poll() soon again".

Thus the newer versions of Kafka decouple polling functionality and heartbeat functionality. So now we have two threads running, the heartbeat thread and the processing thread (polling thread).
The heartbeat frequency is defined by the session.timeout.ms property (default = 10 secs)

Since there is a separate heartbeat thread now, the authors of Kafka Consumer decided to set the default for the polling timeout as INTEGER_MAX. (attribute: max.poll.interval.ms)
Hence no matter how long the processing takes (on the processing/polling thread), the Kafka broker will never consider the consumer to be dead. Only if no poll() request is received after INTERGER_MAX time, then the consumer would be considered dead.
.
Caveat: If your processing has a bug - (e.g. infinite loop, processing has called a third-party webservice and is stuck, etc.), then the consumer will never be pronounced dead and the messages will start getting piled up in that partition. Hence, it may be a good idea to set a realistic time for the polling() interval, so that we can rebalance the messages to other consumers. 

The following 2 stackoverflow discussions were extremely beneficial to us to help us understand the above.

https://stackoverflow.com/questions/47906485/max-poll-intervals-ms-set-to-int-max-by-default
https://stackoverflow.com/questions/39730126/difference-between-session-timeout-ms-and-max-poll-interval-ms-for-kafka-0-10-0

Ruminating on Kafka consumer parallelism

Many developers struggle to understand the nuances of parallelism in Kafka. So jotting down a few points that should help from the Kafka documentation site.

• Consumers label themselves with a consumer group name, and each record published to a topic is delivered to one consumer instance within each subscribing consumer group. Consumer instances can be in separate processes or on separate machines.
• Publishers can publish events into different partitions of Kafka. The producer is responsible for choosing which record to assign to which partition within the topic. This can be done in a round-robin fashion simply to balance load or it can be done according to some semantic partition function (say based on some key in the record).
• The partitions in the log serve several purposes. First, they allow the log to scale beyond a size that will fit on a single server. Each individual partition must fit on the servers that host it, but a topic may have many partitions so it can handle an arbitrary amount of data. Second they act as the unit of parallelism. 

Unlike other messaging middleware, parallel consumption of messages (aka load-balanced consumers) in Kafka is ONLY POSSIBLE using partitions. 

Kafka keeps one offset per [consumer-group, topic, partition]. Hence there cannot be more consumer instances within a single consumer group than there are partitions. 

So if you have only one partition, you can have only one consumer (within a particular consumer-group). You can of-course have consumers across different consumer-groups, but then the messages would be duplicated and not load-balanced. 

Creating a RabbitMQ pipeline using listeners/publishers

In Event Driven Architectures, you often have to create a pipeline of event processing. One of my teams was using Spring AMQP libary and wanted to implement the following basic steps.
1. Read a message from the queue using a RabbitMQ channel.
2. Do some processing and transform the message.
3. Publish the message downstream with the same channel

The sample code given below will help developers in implementing this.

Another scenario is when you have multi-threaded code that is publishing messages to RabbitMQ, then you can use the RabbitMQTemplate with Channel caching. Sample code given below.

Identifying RabbitMQ consumers

One of the challenges my team was facing was to accurately identify the consumers of a RabbitMQ queue. Let's say if you have 5 consumers and you want to kill one of them through the dashboard, you need to identify the consumer.

Each consumer of RabbitMQ has a tag value that can be used to uniquely identify it. By default, RabbitMQ assigns some random number as the consumer tag, but just by looking at this random string tag there is no way to identify the actual consumer. 

To resolve this, you need to create a ConsumerTagStrategy and associate it with the MessageListenerContainer. Code snippet given below: 

Handling RabbitMQ broker re-connections

The RabbitMQ Java Client Library has default support for auto-recovery (since version 4.0.0). Hence the client will try to recover a broken connection unless you explicitly disable it.

If you want to fine-tune the retry mechanism, then the examples given in the below link would help.
http://www.rabbitmq.com/api-guide.html#recovery

Or alternatively, you can use the super easy Spring RabbitTemplate class that has retry options. The RabbitTemplate class wraps the RabbitMQ Java client and provides all the goodies of Spring. 

Handling MQTT broker re-connections

Any client that connects to an MQTT broker needs the ability to handle a connection failure.

The popular Eclipse Paho library now has support for reconnects as described here - http://www.eclipse.org/paho/files/mqttdoc/MQTTAsync/html/auto_reconnect.html
Sample client code available here - https://github.com/SolaceLabs/mqtt-3.1.1-test

If you want to understand how Paho implements reconnect, then have a look at this source file - https://github.com/eclipse/paho.mqtt.java/blob/master/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java

Alternatively, we can use the Spring Integration Framework that encapsulates the Paho library and provides options to configure connection retry logic. https://docs.spring.io/spring-integration/reference/html/mqtt.html

Ruminating on MQTT load balancing for scalable IoT event processing

MQTT brokers support the publish/subscribe paradigm. But what if you need to scale out the processing of MQTT messages over a cluster of nodes (message consumers)?

In IoT environments, we need to process thousands of events per second and hence need to load-balance incoming messages across multiple processing nodes. 

Unfortunately, the standard MQTT specification does not support this concept.

But many MQTT brokers support this as a non-standard feature - e.g. HiveMQ supports Shared Subscriptions. 

The IBM Watson IoT platform also supports shared subscriptions as described here - https://developer.ibm.com/recipes/tutorials/shared-subscription-in-ibm-iot-foundation/

If you are using the IBM Watson IoT java libraries, you need to "Shared-Subscription” property is set to “true”. If you are using any other client like Eclipse Paho, then you must use the client id of the form “A:org_id:app_id”

Note: Please note the capital 'A' in the client ID. This marks the application as a scalable application for load-balancing. We just changed the small 'a' to a capital 'A' and could load-balance our mqtt consumers. 

Ruminating on AMQP internals and JMS equivalent semantics

Many Java developers often use JMS APIs to communicate with the message broker. The JMS API abstracts away the internal implementation complexities of the message broker and provides a unified interface for the developer.

But if you are interested in understanding the internals of AMQP, then the following old tutorial on the Spring site is still the best - https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-by-rabbitmq/ and https://spring.io/understanding/AMQP

By understanding the core concepts of exchange, queue and binding keys, you can envisage multiple integration patterns such as pub/sub, req/reply, etc.

Jotting down snippets from the above sites on the similarities between JMS and AMQP.

JMS has queues and topics. A message sent on a JMS queue is consumed by no more than one client. A message sent on a JMS topic may be consumed by multiple consumers. AMQP only has queues. AMQP producers don't publish directly to queues. A message is published to an exchange, which through its bindings may get sent to one queue or multiple queues, effectively emulating JMS queues and topics.

AMQP has exchanges, routes, and queues. Messages are first published to exchanges with a routing key. Routes define on which queue(s) to pipe the message. Consumers subscribing to that queue then receive a copy of the message. If more than one consumer subscribes to the same queue, the messages are dispensed in a round-robin fashion.

It is very important to understand the difference between routing key and binding key. Publishers publish messages to the AMQP Exchange by giving a routing key; which is typically in the form of {company}.{product-category}.{appliance-type}.{appliance-id}.....

You create a queue on AMQP by specifying the binding key. For e,g, if your binding key is {company}.{product-category}.# then all messages for that product category would come to this queue.

While creating subscribers, you have two options - either bind to an existing queue (by giving the queue name) or create a subscriber private queue by specifying a binding key. 

Concurrency and scaling strategies for MDPs and MDBs

Message Driven Beans offer a lot of advantages over a standalone JMS consumer as listed in this blog-post. The Spring framework provides us with another lightweight alternative called MDP (Message Driven POJOs) that has all the goodies of MDB, but without any heavy JEE server baggage.

The strategy for implementing a scalable and load-balanced message consumption solution is a bit different in MDBs vs. MDPs.

An MDB is managed by the JEE container and is 'thread-safe' by default. The JEE container maintains a pool of MDBs and allows only one thread to execute an MDB at one time. Thus if you configure your JEE container to spawn 10 MDBs, you can have 10 JMS consumers processing messages concurrently.

Spring MDPs are typically managed by the DMLC (DefaultMessageListenerContainer). Each MDP is typically a singleton, but can have multiple threads running through it. Hence an MDP is NOT 'thread-safe' by default and we have to make sure that our MDPs are stateless - e.g. do not have instance variables, etc. The DMLC can be configured for min/max concurrent consumers to dynamically scale the number of consumers. All connection, session and consumer objects are cached by Spring to improve performance. Jotting down some important stuff to remember regarding DMLC from the Spring Java Docs.

"On startup, DMLC obtains a fixed number of JMS Sessions to invoke the listener, and optionally allows for dynamic adaptation at runtime (up to a maximum number). Actual MessageListener execution happens in asynchronous work units which are created through Spring's TaskExecutor abstraction. By default, the specified number of invoker tasks will be created on startup, according to the "concurrentConsumers" setting."

It is also possible in Spring to create a pool of MDPs and give one to each TaskExecutor, but we have never tested this and never felt the need for this. Making your MDPs stateless and hence 'thread-safe' would suffice for almost all business needs. Nevertheless, if you want to create a pool of MDP objects, then this link would help - http://forum.spring.io/forum/spring-projects/integration/jms/25794-mdb-vs-mdp-concurrency

Another good article on Spring MDP scaling that is worth perusing is here - http://bsnyderblog.blogspot.in/2010/05/tuning-jms-message-consumption-in.html

Implementing a Request Response with Messaging (JMS)

Quite often, we need to implement a request/response paradigm on JMS - e.g. calling a backend function on a mainframe or a third-party interface that has an AMQP endpoint. To implement this, we have the following 3 options:

Option 1: The client creates a temporary queue and embeds the name-address of this queue in the message header (JMSReplyTo message header) before sending it to the server. The server processes the request message and writes the response message on to the temp queue mentioned in the JMSReplyTo header.
The advantage of this approach is that the server does not need to know the destination of the client response message in advance.
An important point to remember is that a temporary queue is only valid till the connection object is open. Temporary queues are generally light-weight, but it depends on the implementation of the MOM (message-oriented middleware).

 Option 2: Create a permanent queue for response messages from the server. The client sets a correlation ID on the message before it is sent to the server. The client then listens on the response queue using a JMS selector - using the correlation ID header value as the selector property. This ensures that only the appropriate message from the response queue is delivered to the appropriate client.

Option 3: Use a combination of Option 1 and Option 2.  Each client creates a temporary queue and JMS Consumer on startup. We need to use this option if the client does not block for each request. So the client can keep on sending messages and listen to multiple response messages on the temp queue and then match the req/res using the correlation ID.

Spring has implemented this design pattern in the JMS Outbound Gateway class - http://docs.spring.io/spring-integration/reference/html/jms.html#jms-outbound-gateway

The JMS spec also contains an API for basic req/res using temporary queues - QueueRequestor and TopicRequestor. So if your MOM supports JMS, then you can use this basic implementation if it suffices your needs.

The following links would serve as a good read on this topic.
http://www.enterpriseintegrationpatterns.com/patterns/messaging/RequestReplyJmsExample.html
http://activemq.apache.org/how-should-i-implement-request-response-with-jms.html

Exploring Apache Kafka..

We had successfully used ActiveMQ and RabbitMQ in many projects and never felt the need to explore any other message broker. Today, my colleague introduced me to 'Apache Kafka' and was drooling over the high performance and reliability it provided. Kafka is extensively used within LinkedIn and can be used in many use-cases.

The following blog post gives a good performance benchmark of Kafka.
http://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines

Another good blog post worth reading is: http://engineering.linkedin.com/distributed-systems/log-what-every-software-engineer-should-know-about-real-time-datas-unifying

Another good tutorial on using Kafka to push messages to Hadoop is available here - http://hortonworks.com/hadoop-tutorial/simulating-transporting-realtime-events-stream-apache-kafka/