If you are planning to deploy Fuse ESB in a Kubernetes cluster, you may need to integrate it with a message broker. (For more details on deploying Red Hat JBoss Fuse, read our blog, Deploying Red Hat JBoss Fuse using Azure Container Services and Kubernetes.) Your choice of message broker platform and deployment configuration will depend on several requirements. Here are some requirements that typically impact your technology stack selection:

Need of persistent messaging and durable delivery Specific requirements on high availability (HA)/failover Low-latency delivery Velocity (number of messages, e.g. > 1 M/sec) Volume (message size, e.g. > 1 Mbyte)

If requirements 3 through 5 are most critical for you (for example, you need real-time processing or big data handling), an independent network of brokers may be a better option than a cluster.

For this tutorial, let’s assume we simply need durable messaging and a simple infrastructure to process a low volume of messages. In this case, the obvious solution is to turn on the embedded ActiveMQ in Red Hat Fuse and configure it to use an external database.

In the event of failure when using this solution, one of the standby nodes is automatically activated. This ensures no messages are lost because all of the messages always persist in the database outside of the cluster.

The pros with this solution:

This is a simple deployment into an existing Red Hat Fuse installation.

You have simple failover scenario options such as a self-healing cluster and configuration of a redundant message database. Alternately, you can just rely on database backups.

The cons with this solution:

Using an external database decreases performance.

A high number or large size of messages may lead to extensive memory consumption of the Message Broker, which can kill the Red Hat Fuse node.

For our tutorial, this solution nicely fits our needs. To get started, we’ll introduce changes on three layers:

ActiveMQ configuration; Red Hat Fuse configuration (a Docker file to create the Docker Image with Fuse installed and configured) Kubernetes configuration.

Want a stress-free K8S cluster management experience? Kublr can help. Download our demo, Kublr-in-a-Box.

Configuring ActiveMQ

Change the persistence configuration by defining the JDBC persistence adapter to connect to the MS SQL Server database. After this change, we’ll have a master slave cluster configuration in which the first node that starts will be the ActiveMQ. This node will lock the database, while the other nodes will be in standby mode:

sed -i "s/<\/beans>/$(sed -e 's/[\&/]/\\&/g' -e 's/$/\

/' persistence.txt | tr -d '

')<\/beans>/" etc/activemq.xml sed -i "s/<kahaDB.*/<jdbcPersistenceAdapter dataDirectory=\"..\/activemq-data\" dataSource=\"#mssql-ds\" ><adapter><transact-jdbc-adapter\/><\/adapter><\/jdbcPersistenceAdapter>/" etc/activemq.xml

Where “persistence.txt” contains:

<bean id="mssql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">

<property name="driverClassName" value="com.microsoft.sqlserver.jdbc.SQLServerDriver"/>

<property name="url" value="jdbc:sqlserver://HOST:1433;databaseName=DATABASE" />

<property name="username" value="****"/>

<property name="password" value="****"/>

</bean>

Next, reconfigure the ActiveMQ to enable the message scheduling and bind the nodes to all network interfaces:

sed -i 's/<broker/<broker schedulerSupport="true" /' etc/activemq.xml

sed -i 's/tcp:\/\/${bindAddress}:${bindPort}/tcp:\/\/0.0.0.0:61616/' etc/activemq.xml

Configuring Red Hat Fuse

To configure Red Hat Fuse, install the “activemq-camel” feature and several libraries. In order to do this, add the following commands to the Red Hat Fuse Docker file as described in this post on Deploying Red Hat JBoss Fuse using Azure Container Services:

bin/client 'osgi:install -s mvn:commons-pool/commons-pool/1.6'

bin/client 'osgi:install -s mvn:commons-dbcp/commons-dbcp/1.4'

bin/client 'osgi:install -s mvn:com.microsoft.sqlserver/mssql-jdbc/6.2.0.jre8'



bin/client features:install activeq-camel

The complete Docker file for Red Hat Fuse with ActiveMQ is:

# Use latest jboss/base-jdk:8 image as the base

FROM jboss/base-jdk:8 MAINTAINER Evgeny Pishnyuk < maintainer-email@gmail.com

ENV DEPLOY_CLOUD_STORAGE= ENV DEPLOY_LOCAL_STORAGE=installENV DEPLOY_CLOUD_STORAGE= https://your-cloud-storage-with-prepared-artifacts ENV FUSE_VERSION 6.3.0.redhat-262 RUN curl $DEPLOY_CLOUD_STORAGE/jboss-fuse-karaf-$FUSE_VERSION.zip > /opt/jboss/jboss-fuse-karaf.zip

WORKDIR /opt/jboss

RUN unzip jboss-fuse-karaf.zip -d /opt/jboss && rm *.zip

RUN ln -s "jboss-fuse-$FUSE_VERSION" jboss-fuse # We turn on the default admin user. Please consider password

RUN sed -i 's/#admin/admin/' etc/users.properties # We install components that we needed

RUN bin/fuse server & \

sleep 30 && \

bin/client log:clear && \

bin/client 'osgi:install -s mvn:xom/xom/1.2.5' && \

bin/client features:install camel-jetty && \

bin/client features:install camel-xmljson && \

bin/client 'osgi:install -s mvn:commons-pool/commons-pool/1.6' && \

bin/client 'osgi:install -s mvn:commons-dbcp/commons-dbcp/1.4' && \

bin/client 'osgi:install -s mvn:com.microsoft.sqlserver/mssql-jdbc/6.2.0.jre8' && \

bin/client features:install activeq-camel && \

sleep 10 && \

bin/client log:display && \

bin/client 'shutdown -f' && \

sleep 5 # !Usually it is more affordable to use inheritance of Docker Containers and here will be a split point WORKDIR /opt/jboss/jboss-fuse COPY $DEPLOY_LOCAL_STORAGE/*.jar /opt/deploy/ # We deploy our service – we do it in different step to save time for building of Docker Image RUN bin/fuse server & \

sleep 30 && \

bin/client log:clear && \

bin/client 'osgi:install -s file:/opt/deploy/some-service.jar' && \

sleep 10 && \

bin/client log:display && \

bin/client 'shutdown -f' && \

sleep 5 # Add ports of your services

EXPOSE 8181 8101 1099 44444 61616 1883 5672 61613 61617 8883 5671 61614 CMD /opt/jboss/jboss-fuse/bin/fuse server

Configuring Kubernetes

Start the Red Hat Fuse in a Kubernetes StatefulSet to get predictable DNS-names for our nodes. The names will follow this pattern:

{StatefulSet.name}-{replica_no}.{Headless.name}

In our example, the names are: rhesb-0.activemq and rhesb-1.activemq.

apiVersion: apps/v1beta1

kind: StatefulSet

metadata:

name: rhesb

spec:

serviceName: "activemq"

replicas: 2

template:

metadata:

labels:

name: rhesb

app: rhesb

spec:

terminationGracePeriodSeconds: 5

imagePullSecrets:

- name: regsecret

containers:

- name: rhesb

imagePullPolicy: Always

image: xxx.azurecr.io/rhesb:latest

ports:

- containerPort: 8051

… other needed ports

readinessProbe:

tcpSocket:

port: 8181

initialDelaySeconds: 60

periodSeconds: 20

The Kubernetes Headless service will resolve node names via the DNS. The Headless service name must match the ServiceName in StatefulSet.

apiVersion: v1

kind: Service

metadata:

name: activemq

spec:

ports:

- port: 61616

protocol: TCP

targetPort: 61616

selector:

name: rhesb

clusterIP: None

Usage Example

In the configuration fragment of Camel (a framework under Red Hat Fuse), we define the ActiveMQ connection properties in the bean with the ID “activemq” and implement two Camel Routes as shown below which will move a message between two queues back and forth every minute.

Note:

Replace the localhost in failover:(tcp://localhost:61616) with the list of actual nodes during the container startup.

with the list of actual nodes during the container startup. Camel <from> URI contains ?acknowledgementModeName=CLIENT_ACKNOWLEDGE indicates message delivery was acknowledged after route is completed without exceptions.



<property name="connectionFactory">

<bean class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="failover:(tcp://localhost:61616)?initialReconnectDelay=100" />

<property name="userName" value="admin"/>

<property name="password" value="admin"/>

</bean>

</property>

</bean>

<camelContext id="MockPulseAPIContext" xmlns="

<route id="poc_activemqincoming">

<from id="qfrom" uri="activemq:queue:incoming?acknowledgementModeName=CLIENT_ACKNOWLEDGE"/>

<to id="qto" uri="activemq:queue:sleep1min"/>

<log id="_logq" message="Incoming ${body}"/>

</route>



<route id="poc_activemqdelayed">

<from id="qfromdelayed" uri="activemq:queue:sleep1min?acknowledgementModeName=CLIENT_ACKNOWLEDGE"/>

<delay asyncDelayed="true" id="qdelay">

<constant>60000</constant>

</delay>

<to id="qtodelayed" uri="activemq:queue:incoming"/>

</route>

</camelContext> http://camel.apache.org/schema/blueprint "> 60000

Testing

To test your configuration, send a “Something for the Masses” message, and confirm that the message was processed on both nodes.

In case the ActiveMQ master node shuts down or fails, one of the nodes in standby mode will be promoted to become the master, so no message from the ActiveMQ queues will be lost.

Share your thoughts and questions in the comments section below.

Need a user-friendly tool to set up and manage your K8S cluster? Check out Kublr-in-a-Box. To learn more, visit kublr.com.