Background
Many customers desire to send events from KurrentDB to Azure Event Hubs for dissemination to downstream applications and technologies. Since Azure Event Hubs provides a Kafka endpoint, integration with KurrentDB can easily be configured using our Kafka Connector Sink. This document outlines the steps necessary, and where to find the required information in the Azure control panel
NOTE: This document assumes you are familiar with connectors, and how to write regular expressions to filter the events sent to the connector sink. More information on these topics can be found in the documentation linked above
Gathering Information
The following information is required and can be gathered from your Azure Event Hubs portal:
IMPORTANT NOTES:
- Your Azure Event Hub namespace configuration must be running at the Standard or higher Pricing Tier to expose the Kafka API option. See this link for more information - Azure Event Hubs for Apache Kafka
- Azure limits the transaction throughput (append rate) through Transaction Units in the Standard Pricing Tier. The default configuration is 1 TU, to a maximum of 40 TU. Azure limits the transaction throughput (append rate) through Processing Units in the Premium Pricing Tier. The default configuration is 1 PU to a maximum of 16 PU. To avoid throttling of events sent to Azure Event Hubs, customers should consider using Auto-inflate, or set appropriate TU / PU provisioning in Azure. See this link for more information - Scaling with Event Hubs - Azure
Host Name
The Host name field is used as the bootstrap server address in KurrentDB's connector configuration. It can be found on the Event Hubs Namespace configuration page for your desired Event Hub
In this case, tonybus.servicebus.windows.net
Password
The Password is used for authentication between KurrentDB's connector subsystem and Azure Event Hubs. It is the field value of Primary Connection String. This value can be found under your Event Hubs Namespace → Settings → Shared access policies
Choose your desired Access Policy and click its name in the list. Click the Copy to Clipboard button next to the Primary connection string field
In this case, Endpoint=sb://tonybus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXX
NOTE: Other authentication mechanisms may be supported depending on your configuration and organizational policy. See the above links for more information
Configuring KurrentDB
When working with Usernames and Passwords, KurrentDB's connector subsystem will store these in an encrypted format inside configuration streams. To prepare the server to encrypt these items, customers must set an additional configuration value:
Connectors:
…
DataProtection:
Token: "12345678"
This configuration parameter may also be set with an environment variable such as KURRENTDB_CONNECTORS__DATAPROTECTION__TOKEN="12345678"
In this case, the token of 12345678 will be used, and must never be changed, or all previous configuration that is encrypted may no longer be decrypted
NOTE: Configuration may also be completed with a tokenfile. See the documentation linked above for more information
Creating the Connector
Connectors are created by forming a JSON configuration payload, and building a CURL command. To build a JSON payload, you should determine:
- Topic: To what destination topic do you wish to append events in your Azure Event Hub. The topic will be created automatically in your Azure Event Hubs namespace if it does not already exist
- Bootstrap Servers: This is your Host name gathered above with a port of 9093
- Password: this is the Primary connection string gathered above
Here is an example JSON payload with reference to the Azure Event Hubs properties fetched earlier:
{
"settings": {
"InstanceTypeName": "kafka-sink",
"topic": "CDC",
"bootstrapServers": "tonybus.servicebus.windows.net:9093",
"Subscription:Filter:Scope": "Stream",
"Subscription:Filter:Expression": "^cdc.bronze.\*?",
"subscription:initialPosition": "earliest",
"authentication:username": "$ConnectionString",
"authentication:password": "Endpoint=sb://tonybus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXX",
"authentication:securityProtocol": "SaslSsl",
"authentication:saslMechanism": "Plain",
"waitForBrokerAck": "true"
}
}
NOTE: The hard settings of securityProtocol, saslMechanism, username, and InstanceTypeName apply to any connection to Azure Event Hubs. Other fields such as Subscription.*, waitForBrokerAck, etc. may be customized to your liking according to the Connectors documentation
Here is an example shell script of the CURL commands with embedded JSON payload to create and start the connector:
#!/bin/bash
curl -v -i -k \
-X POST https://localhost:2113/connectors/eventhubs-demo \
-H "Content-Type: application/json" \
-u "admin:changeit" \
-d '{
"settings": {
"InstanceTypeName": "kafka-sink",
"topic": "CDC",
"bootstrapServers": "tonybus.servicebus.windows.net:9093",
"Subscription:Filter:Scope": "Stream",
"Subscription:Filter:Expression": "^cdc.bronze.*?",
"subscription:initialPosition": "earliest",
"authentication:username": "$ConnectionString",
"authentication:password": "Endpoint=sb://tonybus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXX",
"authentication:securityProtocol": "SaslSsl",
"authentication:saslMechanism": "Plain",
"waitForBrokerAck": "true"
}
}' \
sleep 10
curl -v -i -k -X POST https://localhost:2113/connectors/eventhubs-demo/start -u "admin:changeit"
Here is a sample shell script to stop and delete the connector:
#!/bin/bash
curl -v -i -k -X POST https://localhost:2113/connectors/eventhubs-demo/stop -u "admin:changeit"
sleep 10
curl -v -i -k -X DELETE https://localhost:2113/connectors/eventhubs-demo -u "admin:changeit"
Conclusion
In this document we have detailed the process to collect the necessary information to create a Kafka connector to Azure Event Hubs. For any additional information, please review the linked documentation. For help, please connect with us on Discord, Stack Overflow, Discuss, or directly with your Field Technical Services team member
