How to monitor the Replication Queues through Java

This post will explain the approach to monitor the Replication Queues of Adobe Experience Manager(AEM) through java

Enable the Remote JMX in the AEM server:
Add the following configurations as part of CQ_JVM_OPTS in AEM startup file (start.sh or start.bat)

-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=50055

e.g.

CQ_JVM_OPTS='-server -Xmx6144m -XX:MaxPermSize=512M -Djava.awt.headless=true -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=50055'

Restart the server.

Configuration.txt:
ServerHost=xxxxxxxxxxx
ServerPort=50055
QueueNames=publish,publish_Dev
Environment=Development
EmailHost=xxxxxx
Email_To=xxxxxx
Email_From=xxxxxxx
QueueThreshold=15

GetReplicationQueueStatus.java
import java.io.*;
import java.util.*;
import javax.management.*;
import javax.management.remote.*;
import javax.mail.*;
import javax.mail.internet.*;

public class GetReplicationQueueStatus {
static JMXConnector m_connector=null;
public static void main(String[] args) {
FileInputStream in=null;
try
{
Properties configuration = new Properties();
in= new FileInputStream("Configuration.txt");
configuration.load(in);
String serverHost=configuration.getProperty("ServerHost");

String serverUrl = "service:jmx:rmi:///jndi/rmi://"+serverHost+":"+configuration.getProperty("ServerPort")+"/jmxrmi";

JMXServiceURL serviceURL = new JMXServiceURL(serverUrl);
m_connector =JMXConnectorFactory.newJMXConnector(serviceURL,null);
m_connector.connect();
MBeanServerConnection m_connection = m_connector.getMBeanServerConnection();
Set<ObjectName> queryResult =m_connection.queryNames(new ObjectName("com.adobe.granite.replication:type=agent,*"),null);
String queueNames=configuration.getProperty("QueueNames");
List<String> queueNameList = Arrays.asList(queueNames.split(","));
for(ObjectName objectName : queryResult) {
if(queueNameList.contains(objectName.getKeyProperty("id").replaceAll("\"", "")))
{
String attrNames[] =
{ "Enabled",
"QueueNumEntries",
"QueueBlocked",
"QueuePaused"
};

AttributeList attrList = m_connection.getAttributes(objectName, attrNames);
String isEnabled= ((Object) attrList.get(0)).toString().split("=")[1];
String queueNumEntries= ((Object) attrList.get(1)).toString().split("=")[1];
String isQueueBlocked= ((Object) attrList.get(2)).toString().split("=")[1];
String isQueuePaused= ((Object) attrList.get(3)).toString().split("=")[1];
System.out.println( "Values: " +isEnabled+" "+ queueNumEntries+" "+isQueueBlocked+" "+ isQueuePaused);
String to=configuration.getProperty("Email_To");
String from=configuration.getProperty("Email_From");
String emailHost=configuration.getProperty("EmailHost");
String queueThreshold=configuration.getProperty("QueueThreshold");
String environment=configuration.getProperty("Environment");
if(!isEnabled.trim().equals("true"))
{
String subject=environment+": Replication Queue "+"\""+objectName.getKeyProperty("id").replaceAll("\"", "")+"\""+" is Disabled";
sendEmail(to,from, subject,subject+"\n"+"Server: "+serverHost, emailHost);
}else if(isQueueBlocked.trim().equals("true"))
{
String subject=environment+": Replication Queue "+"\""+objectName.getKeyProperty("id").replaceAll("\"", "")+"\""+" is Blocked";
sendEmail(to,from, subject,subject+"\n"+"Server: "+serverHost, emailHost);
}else if(isQueuePaused.trim().equals("true"))
{
String subject=environment+": Replication Queue "+"\""+objectName.getKeyProperty("id").replaceAll("\"", "")+"\""+" is Paused";
sendEmail(to,from, subject,subject+"\n"+"Server: "+serverHost, emailHost);
}else if(Integer.parseInt(queueNumEntries.trim())>Integer.parseInt(queueThreshold.trim()))
{
String subject=environment+": Replication Queue "+"\""+objectName.getKeyProperty("id").replaceAll("\"", "")+"\""+" is Queued with "+queueNumEntries+" Requests";
sendEmail(to,from, subject,subject+"\n"+"Server: "+serverHost, emailHost);
}

}

}
}catch(Exception e)
{
e.printStackTrace();
}finally {
try {
if(m_connector!=null)
{
m_connector.close();
}
if(in!=null)
{
in.close();
}
}catch(Exception e)
{

}
}

}

static void sendEmail(String to,String from,String subject,String body,String host)
{
Properties properties = System.getProperties();
properties.setProperty("mail.smtp.host", host);
Session session = Session.getDefaultInstance(properties);

try {
MimeMessage message = new MimeMessage(session);
message.setFrom(new InternetAddress(from));
message.addRecipient(Message.RecipientType.TO, new InternetAddress(to));
message.setSubject(subject);
message.setText("Hi Team\n\n"+body+"\n\n"+"Regards\nAdmin");
Transport.send(message);
System.out.println("Email send successfully...");

}catch (MessagingException mex) {
mex.printStackTrace();
}
}

}

Configure the data in Configuration.txt file accordingly.
An email will be triggered whenever the configured Queue is disabled, blocked, paused or the pending request is more than the configured threshold.

The executable jar file can be generated for the java class and scheduled for continuous monitoring.