Export JMX metrics from Kafka connectors in Amazon Managed Streaming for Apache Kafka Join with a {custom} plugin

Organizations use streaming purposes to course of and analyze knowledge in actual time and undertake the Amazon MSK Join characteristic of Amazon Managed Streaming for Apache Kafka (Amazon MSK) to run totally managed Kafka Join workloads on AWS. Message brokers like Apache Kafka enable purposes to deal with massive volumes and numerous kinds of knowledge effectively and allow well timed decision-making and on the spot insights. It’s essential to observe the efficiency and well being of every element to assist make sure the seamless operation of information streaming pipelines.

Amazon MSK is a completely managed service that simplifies the deployment and operation of Apache Kafka clusters on AWS. It simplifies constructing and working purposes that use Apache Kafka to course of streaming knowledge. Amazon MSK Join simplifies the deployment, monitoring, and automated scaling of connectors that switch knowledge between Apache Kafka clusters and exterior techniques comparable to databases, file techniques, and search indices. Amazon MSK Join is totally appropriate with Kafka Join and helps Amazon MSK, Apache Kafka, and Apache Kafka appropriate clusters. Amazon MSK Join makes use of a {custom} plugin because the container for connector implementation logic.

Customized MSK join plugins use Java Administration Extensions (JMX) to reveal runtime metrics. Whereas Amazon MSK Join sends a set of join metrics to Amazon CloudWatch, it at the moment doesn’t assist exporting the JMX metrics emitted by the connector plugins natively. These metrics may be exported by modifying the {custom} join plugin code straight, but it surely requires upkeep overhead as a result of the plugin code must be modified each time it’s up to date. On this put up, we reveal an optimum method by extending a {custom} join plugin with extra modules to export JMX metrics and publish them to CloudWatch as {custom} metrics. These extra JMX metrics emitted by the {custom} connectors present wealthy insights into their efficiency and well being of the connectors. On this put up, we reveal how one can export the JMX metrics for Debezium connector when used with MSK Join.

Understanding JMX

Earlier than we dive deep into exporting JMX metrics, let’s perceive how JMX works. JMX is a expertise that you should use to observe and handle Java purposes. Key parts concerned in JMX monitoring are:

• Managed beans (MBeans) are Java objects that symbolize the metrics of the Java utility being monitored. They comprise the precise knowledge factors of the assets being monitored.
• JMX server creates and registers the MBeans with the PlatformMBeanServer. The Java utility that’s being monitored acts because the JMX server and exposes the MBeans.
• MBeanServer or JMX registry is the central registry that retains monitor of all of the registered MBeans within the JMX server. It’s the entry level for all of the MBeans throughout the Java digital machine (JVM).
• JMXConnectorServer acts as a bridge between the JMX shopper and the JMX server and allows distant entry to the uncovered MBeans. JMXConnectorServerFactory creates and manages the JMXConnectorServer. It permits for the customization of the server’s properties and makes use of the JMXServiceURL to outline the endpoint the place the JMX shopper can hook up with the JMX server.
• JMXServiceURL supplies the required data such because the protocol, host, and port for the shopper to hook up with the JMX server and entry the specified MBeans.
• JMX shopper is an exterior utility or instrument that hook up with the JMX server to entry and monitor the uncovered metrics.

JMX monitoring entails the steps proven within the following determine:

JMX monitoring steps embody:

1. The Java utility performing because the JMX server creates and configures MBeans for the specified metrics.
2. JMX server registers the MBeans with the JMX registry.
3. JMXConnectorServerFactory creates the JMXConnectorServer that defines the JMXServiceURL that gives the entry level particulars for the JMX shopper.
4. JMXClient connects to the JMX registry within the JMX server utilizing the JMXServiceURL and the JMXConnectorServer.
5. The JMX server handles shopper requests, interacting with the JMX registry to retrieve the MBean knowledge.

Answer overview

This methodology of wrapping supported Kafka connectors with {custom} code that exposes connector-specific operational metrics allows groups to get higher insights by correlating varied connector metrics with cloud-centered metrics in monitoring techniques comparable to Amazon CloudWatch. This method allows constant monitoring throughout completely different parts of the change knowledge seize (CDC) pipeline, in the end feeding metrics into unified dashboards whereas respecting every connector’s architectural philosophy. The consolidated metrics may be delivered to CloudWatch or the monitoring instrument of your alternative together with associate particular utility efficiency administration (APM) instruments comparable to Datadog, New Relic, and so forth.

We have now the working implementation of this identical method with two widespread connectors: Debezium supply connector and MongoDB Sink Connector. You could find the Github pattern and able to use plugins constructed for every within the repository. Overview the README file for this practice implementation for extra particulars.

For instance, our {custom} implementation for the MongoDB Sink Connector provides a metrics export layer that calculates essential efficiency indicators comparable to latest-kafka-time-difference-ms – which measures the latency between Kafka message timestamps and connector processing time by subtracting the connector’s present clock time from the final acquired file’s timestamp. This tradition wrapper across the MongoDB Sink Connector allows exporting related JMX metrics and publishing them as {custom} metrics to CloudWatch. We’ve open sourced this resolution on GitHub, together with a ready-to-use plugin and detailed configuration steering within the README.

CDC is the method of figuring out and capturing modifications made in a database and delivering these modifications in actual time to a downstream system. Debezium is an open supply distributed platform constructed on prime of Apache Kafka that gives CDC performance. It supplies a set of connectors to trace and stream modifications from databases to Kafka.

Within the subsequent part, we dive deep into the implementation particulars of how you can export JMX metrics from Debezium MySQL Connector deployed as a {custom} plugin in Amazon MSK Join. The connector plugin takes care of making and configuring the MBeans and registering them with the JMX registry.

The next diagram reveals the workflow of utilizing Debezium MySQL Connector as a {custom} plugin in Amazon MSK Join for CDC from an Amazon Aurora MySQL-Suitable Version knowledge supply.

1. MySQL binary log (binlog) is enabled in Amazon Aurora for MySQL to file all of the operations within the order through which they’re dedicated to the database.
2. The Debezium connector plugin element of the MSK Join {custom} plugin constantly screens the MySQL database, captures the row-level modifications by studying the MySQL bin logs, and streams them as change occasions to Kafka matters in Amazon MSK.
3. We’ll construct a {custom} module to allow JMX monitoring on the Debezium connector. This module will act as a JMX shopper to retrieve the JMX metrics from the connector and publish them as {custom} metrics to CloudWatch.

The Debezium connector supplies three kinds of metrics along with the built-in assist for default Kafka and Kafka Join JMX metrics.

• Snapshot metrics present details about connector operation whereas performing a snapshot.
• Streaming metrics present details about connector operation when the connector is studying the binlog.
• Schema historical past metrics present details about the standing of the connector’s schema historical past.

On this resolution, we export the MilliSecondsBehindSource streaming metrics emitted by the Debezium MySQL connector. This metric supplies the variety of milliseconds that the connector is lagging behind the change occasions within the database.

Stipulations

Following are the conditions you want:

• Entry to the AWS account the place you need to arrange this resolution.
• You have got arrange the supply database and MSK cluster by following this setup directions within the MSK Join workshop.

Create a {custom} plugin

Making a {custom} plugin for Amazon MSK Join for the answer entails the next steps:

1. Create a {custom} module: Create a brand new Maven module or challenge that can comprise your {custom} code to:
   1. Allow JMX monitoring within the connector utility by beginning the JMX server.
   2. Create a Distant Methodology Invocation (RMI) registry to allow the entry to the JMX metrics to the purchasers.
   3. Create a JMX metrics exporter to question the JMX metrics by connecting to the JMX server and push the metrics to CloudWatch as {custom} metrics.
   4. Schedule to run the JMX metrics exporter at a configured interval.
2. Package deal and deploy the {custom} module as an MSK Join {custom} plugin.
3. Create a connector utilizing the {custom} plugin to seize CDC from the supply, stream it and validate the metrics in Amazon CloudWatch.

This tradition module extends the connector performance to export the JMX metrics with out requiring any modifications within the underlying connector implementation. This helps make sure that upgrading the {custom} plugin requires solely upgrading the plugin model within the pom.xml of the {custom} module.

Let’s deep dive and perceive the implementation of every step talked about above.

1. Create a {custom} module

Create a brand new Maven challenge with dependencies on Debezium MySQL Connector to allow JMX monitoring, Kafka Join API for configuration, and CloudWatch AWS SDK to push the metrics to CloudWatch.
Arrange a JMX connector server to allow JMX monitoring: To allow JMX monitoring, the JMX server must be began on the time of initializing the connector. That is normally accomplished by setting the setting variables with JMX choices as described in Monitoring Debezium. Within the case of an Amazon MSK Join {custom} plugin, JMX monitoring is enabled programmatically on the time of connector plugin initialization. To attain this:

• Lengthen the MySqlConnector class and override the begin which is the connector’s entry level to execute {custom} code.

public class DebeziumMySqlMetricsConnector extends MySqlConnector{
@Override
	public void begin(Map props) {

• Within the begin methodology of the {custom} connector class (DebeziumMySqlMetricsConnector) that we’re creating, set the next parameters to permit customization of the JMX Server properties by retrieving connector configuration from a config file.

join.jmx.port – The port quantity on which the RMI registry must be created. JMXConnectorServer would take heed to the incoming connections on this port.

database.server.identify – Title of the database that’s the supply for the CDC.

It additionally retrieves the CloudWatch configuration associated properties that can be used whereas pushing the JMX metrics to CloudWatch.

cloudwatch.namespace.identify – CloudWatch NameSpace to which the metrics have to be pushed as {custom} metrics

cloudwatch.area – CloudWatch Area the place the {custom} namespace is created in your AWS account

connectJMXPort = Integer.parseInt(props.getOrDefault(CONNECT_JMX_PORT_KEY, String.valueOf(DEFAULT_JMX_PORT)));
databaseServerName = props.getOrDefault(DATABASE_SERVER_NAME_KEY, "");
cwNameSpace = props.getOrDefault(CW_NAMESPACE_KEY, DEFAULT_CW_NAMESPACE);
cwRegion = props.getOrDefault(CW_REGION_KEY, null);

• Create an RMI registry on the required port (connectJMXPort). This registry is utilized by the JMXConnectorServer to retailer the RMI objects akin to the MBeans within the JMX registry. This enables the JMX purchasers to search for and entry the MBeans on the PlatformMBeanServer.

LocateRegistry.createRegistry(connectJMXPort);

• Retrieve the PlatformMBeanServer and assemble the JMXServiceURL which is within the format service:jmx:rmi://localhost/jndi/rmi://localhost:>/jmxrmi. Create a brand new JMXConnectorServer occasion utilizing the JMXConnectorServerFactory and the JMXServiceURL and begin the JMXConnectorServer occasion.

MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
String jmxServiceURL = String.format(JMX_URL_TEMPLATE, connectJMXPort);
JMXServiceURL url = new JMXServiceURL(jmxServiceURL);
JMXConnectorServer svr = JMXConnectorServerFactory.newJMXConnectorServer(url, null, mbs);
svr.begin();

Implement JMX metrics exporter: Create a JMX shopper to hook up with the JMX server, question the MilliSecondBehindSource metric from the JMX server, convert it into the required format, and export it to CloudWatch.

• Hook up with the JMX Server utilizing the JMXConnectorFactory and JMXServiceURL

JMXServiceURL jmxUrl = new JMXServiceURL(String.format(JMX_URL_TEMPLATE,DebeziumMySqlMetricsConnector.getConnectJMXPort()));
JMXConnector jmxConnector = JMXConnectorFactory.join(jmxUrl, null);
jmxConnector.join();

• Question the MBean object that holds the corresponding metric, for instance, MilliSecondsBehindSource, and retrieve the metric worth utilizing pattern code supplied in msk-connect-custom-plugin-jmx. (you’ll be able to select a number of metrics).
• Schedule the execution of your JMX metrics exporter at common intervals.

getScheduler().schedule(new JMXMetricsExporter(), SCHEDULER_INITIAL_DELAY, SCHEDULER_PERIOD);

Export metrics to CloudWatch: Implement the logic to push related JMX metrics to CloudWatch. You should use the AWS SDK for Java to work together with the CloudWatch PutMetricData API or use the CloudWatch Logs subscription filter to ingest the metrics from a devoted Kafka matter.

Dimension dimension = Dimension.builder()
.identify("DBServerName")
.worth(DebeziumMySqlMetricsConnector.getDatabaseServerName())
.construct();
MetricDatum datum = MetricDatum.builder()
	     .metricName("MilliSecondsBehindSource")
	     .unit(StandardUnit.NONE)
	     .worth(Double.valueOf(msBehindSource))
	     .timestamp(on the spot)
	     .dimensions(dimension).construct();
PutMetricDataRequest request = PutMetricDataRequest.builder()
	  .namespace(DebeziumMySqlMetricsConnector.getCWNameSpace())
	  .metricData(datum).construct();
cw.putMetricData(request);

For extra data, see the pattern implementation for the {custom} module in aws-samples in GitHub. This pattern additionally supplies {custom} plugins packaged with two completely different variations of Debezium MySQL connector (debezium-connector-mysql-2.5.2.Last-plugin and debezium-connector-mysql-2.7.3.Last-plugin) and the next steps would clarify the steps to construct a {custom} plugin utilizing your {custom} code.

2. Package deal the {custom} module and Debezium MySQL connector as a {custom} plugin

Construct and package deal the Maven challenge with the {custom} code as a JAR file and embody the JAR file within the debezium-connector-mysql-2.5.2.Last-plugin folder downloaded from maven repo. Package deal the up to date debezium-connector-mysql-2.5.2.Last-plugin as a ZIP file (Amazon MSK Join accepts {custom} plugins in ZIP or JAR format). Alternatively, you should use the prebuiltcustom-debezium-mysql-connector-plugin.zip accessible in GitHub.

Select the Debezium connector model (2.5 or 2.7) that matches your requirement.

When it’s a must to improve to a brand new model of the Debezium MySQL connector, you’ll be able to replace the model of the dependency and construct the {custom} module and deploy it. By doing this, you’ll be able to preserve the {custom} plugin with out modifying the unique connector code. The GitHub samples present ready-to-use plugins for 2 Debezium connector variations. Nonetheless, you’ll be able to comply with the identical method to improve to the newest connector model as properly.

Create a {custom} plugin in Amazon MSK

1. When you’ve got arrange your AWS assets by following the Getting Began lab, open Amazon S3 console and find the bucket msk-lab-${ACCOUNT_ID}-plugins-bucket/debezium .
2. Add the {custom} plugin created within the earlier part custom-debezium-mysql-connector-plugin.zip to msk-lab-${ACCOUNT_ID}-plugins-bucket/debezium, as proven within the following determine.

3. Change to the Amazon MSK console and select Customized plugins within the navigation pane. Select Create {custom} plugin and, browse the S3 bucket that you just created above and choose the {custom} plugin ZIP file you simply uploaded.

4. Enter custom-debezium-mysql-connector-plugin for the plugin identify. Optionally, enter an outline and select Create Customized Plugin.

5. After a number of seconds it is best to see the plugin is created and the standing is Energetic.
6. Customise the employee configuration for the connector by following the directions within the Customise employee configuration lab.

3. Create an Amazon MSK connector

The subsequent step is to create an MSK connector.

1. From the MSK part select Connectors, then select Create connector. Select custom-debezium-mysql-connector-plugin from the checklist of Customized plugins, then select Subsequent.

2. Enter custom-debezium-mysql-connector within the Title textbox, and an outline for the connector.

3. Choose the MSKCluster-msk-connect-lab from the listed MSK clusters. From the Authentication dropdown, choose IAM.
4. Copy the next configuration and paste it within the connector configuration textbox.

• Change the , , , and placeholders with the corresponding particulars for the assets in your account.
• Overview the matter.prefix, database.consumer, matter.prefix, database.server.id, database.server.identify, database.port, database.embody.checklistparameters within the configuration. These parameters are configured with the values used within the workshop. Replace them with the main points akin to your configuration if in case you have personalized it in your account.
• Observe that the connector.classparameter is up to date with the certified identify of the subclass of MySqlConnector class that you just created within the {custom} module.
• The join.jmx.portparameter specifies the default port to start out the JMX server. You’ll be able to configure this to any accessible port.

connector.class=com.amazonaws.msk.debezium.mysql.join.DebeziumMySqlMetricsConnector duties.max=1
embody.schema.modifications=true
matter.prefix=salesdb
worth.converter=org.apache.kafka.join.json.JsonConverter
key.converter=org.apache.kafka.join.storage.StringConverter
database.consumer=grasp
database.server.id=123456
database.server.identify=salesdb
database.port=3306
key.converter.schemas.allow=false
database.hostname=
database.password=
worth.converter.schemas.allow=false
database.embody.checklist=salesdb
schema.historical past.inner.kafka.matter=inner.dbhistory.salesdb
schema.historical past.inner.kafka.bootstrap.servers=
schema.historical past.inner.producer.sasl.mechanism=AWS_MSK_IAM
schema.historical past.inner.client.sasl.mechanism=AWS_MSK_IAM
schema.historical past.inner.producer.sasl.jaas.config=software program.amazon.msk.auth.iam.IAMLoginModule required;
schema.historical past.inner.client.sasl.jaas.config=software program.amazon.msk.auth.iam.IAMLoginModule required;
schema.historical past.inner.producer.sasl.shopper.callback.handler.class=software program.amazon.msk.auth.iam.IAMClientCallbackHandler
schema.historical past.inner.client.sasl.shopper.callback.handler.class=software program.amazon.msk.auth.iam.IAMClientCallbackHandler
schema.historical past.inner.client.safety.protocol=SASL_SSL
schema.historical past.inner.producer.safety.protocol=SASL_SSL
join.jmx.port=7098
cloudwatch.namespace.identify=MSK_Connect
cloudwatch.area=

5. Observe the remaining directions from the Create MSK Connector lab and create the connector. Confirm that the connector standing modifications to Working.

Debezium MySQL {custom} connector model (2.7.3) supplies extra flexibility to configure optionally available properties that may be added to your MSK connector configuration and selectively embody and exclude metrics to emit to CloudWatch. The next are the instance configuration properties that can be utilized with model 2.7.3 :

• cloudwatch.debezium.streaming.metrics.embody – A comma-separated checklist of streaming metrics kind that should be exported to CloudWatch as {custom} metrics.
• cloudwatch.debezium.streaming.metrics.exclude – Specify a comma-separated checklist of streaming metrics varieties to exclude from being despatched to CloudWatch as {custom} metrics.
• Equally embody and exclude properties for snapshot metrics kind are cloudwatch.debezium.snapshot.metrics.embody and cloudwatch.debezium.snapshot.metrics.exclude
• Embrace and exclude properties for schema historical past metrics kind are cloudwatch.debezium.schema.historical past.metrics.embody and cloudwatch.debezium.schema.historical past.metrics.exclude

The next is a pattern configuration excerpt.

  "cloudwatch.debezium.streaming.metrics.embody": "LastTransactionId, TotalNumberOfEventsSeen, MilliSecondsBehindSource,CapturedTables",
  "cloudwatch.debezium.streaming.metrics.exclude": "LastTransactionId",
  "cloudwatch.debezium.schema.historical past.metrics.exclude": "MilliSecondsSinceLastAppliedChange",

Overview the GitHub README file for extra particulars on using these properties with MSK connector configurations.

Confirm the replication within the Kafka cluster and CloudWatch metrics

Observe the directions within the Confirm the replication within the Kafka cluster lab to arrange a shopper and make modifications to the supply DB and confirm that the modifications are captured and despatched to Kafka matters by the connector.

To confirm that the connector has revealed the JMX metrics to CloudWatch, go to the CloudWatch console and select Metrics within the navigation pane, then select All Metrics. Below Customized namespace, you’ll be able to see MSK_Connect with database identify because the dimension. Choose the database identify to view the metrics.

Choose the MilliSecondBehindSource metric with statistic as Common within the Graphed Metric to plot the graph. You’ll be able to confirm that the MilliSecondBehindSource metric worth is larger than zero every time any operation is being carried out on the supply database and returns to zero through the idle time.

Clear up

Delete the assets that you just created such because the Aurora DB, Amazon MSK Cluster and connectors by following the directions at Cleanup within the Amazon MSK Join lab if in case you have been following alongside to arrange the answer in your account.

Conclusion

On this put up, we confirmed you how you can lengthen the Debezium MySQL connector plugin with an extra module to export the JMX metrics to CloudWatch as {custom} metrics. As a subsequent step, you’ll be able to create a CloudWatch alarm to observe the metrics and take remediation actions when the alarm is triggered. Along with exporting the JMX metrics to CloudWatch, you’ll be able to export these metrics to third-party purposes comparable to Prometheus or DataDog utilizing CloudWatch Metric Streams. You’ll be able to comply with the same method to export the JMX metrics of different connectors from MSK Join. You’ll be able to study extra about creating your personal connectors by visiting the Connector Developer Information and how you can deploy them as {custom} plugins within the MSK Join documentation.

In regards to the authors

Jaydev Nath is a Options Architect at AWS, the place he works with ISV clients to construct safe, scalable, dependable, and cost-efficient cloud options. He brings robust experience in constructing SaaS structure on AWS with a deal with Generative AI and knowledge analytics applied sciences to assist ship sensible, useful enterprise outcomes for patrons.

David John Chakram is a Principal Options Architect at AWS. He focuses on constructing knowledge platforms and architecting seamless knowledge ecosystems. With a profound ardour for databases, knowledge analytics, and machine studying, he excels at reworking advanced knowledge challenges into modern options and driving companies ahead with data-driven insights.

Sharmila Shanmugam is a Options Architect at Amazon Net Providers. She is enthusiastic about fixing the purchasers’ enterprise challenges with expertise and automation and scale back the operational overhead. In her present position, she helps clients throughout industries of their digital transformation journey and construct safe, scalable, performant and optimized workloads on AWS.