Tutorial: Connect to Azure Event Hubs for Kafka using Go
Tags: tutorial, go, showdev, beginners
Published: 2019-10-09 12:34:16

This blog will demonstrate how to interact with [Event Hubs Kafka](https://docs.microsoft.com/azure/event-hubs/event-hubs-for-kafka-ecosystem-overview?WT.mc_id=devto-blog-abhishgu) cluster using the [Sarama Kafka client](https://github.com/Shopify/sarama) library. The [`sarama` package](https://godoc.org/github.com/Shopify/sarama) provides a pure Go client that supports Kafka v 0.8 and above.

![](https://thepracticaldev.s3.amazonaws.com/i/raxcmg9qlc63eoa3egw7.jpg)

[Azure Event Hubs](https://docs.microsoft.com/azure/event-hubs/?WT.mc_id=devto-blog-abhishgu) is a streaming platform and event ingestion service, capable of receiving and processing millions of events per second. It also provides a Kafka endpoint that can be used by your existing Kafka based applications as an alternative to running your own Kafka cluster. Since Azure Event Hubs exposes a protocol that is binary compatible with Kafka versions 1.0, you can start using the Kafka endpoint from your existing applications with no code change but a minimal configuration change. This also supports frameworks like [Kafka Connect](https://kafka.apache.org/documentation/#connect) (currently in preview), [MirrorMaker](https://kafka.apache.org/documentation/#basic_ops_mirror_maker) etc.

What's covered?

- **Code**: how to configure and use the Sarama Go client to talk to Event Hubs Kafka endpoint and build producer, consumer apps
- **Setup**: use Azure CLI to quickly bootstrap an Event Hubs for Kafka instance
- **Test**: run the producer and consumer app to try the end to end scenario

> as always, the code is available on [GitHub](https://github.com/abhirockzz/eventhubs-kafka-go-sarama)

### Pre-requisites

If you don't have an Azure subscription, just create a [free account](https://azure.microsoft.com/free/?WT.mc_id=devto-blog-abhishgu) and get going! You will also need the [Azure CLI](https://docs.microsoft.com/cli/azure/install-azure-cli?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu). Of course, you will need to have [Go installed](https://golang.org/doc/install) as well.

### Code walkthrough

The app is pretty simple and consists of a producer and a consumer built using the Sarama Go client. Let's skim through the code real quick

![](https://media.giphy.com/media/3oKIPnAiaMCws8nOsE/giphy-downsized.gif)

#### Configuration for connecting to Event Hubs for Kafka

You need to pass a `sarama.Config` object in order to create a producer or consumer instance.

    func getConfig() *sarama.Config {
        config := sarama.NewConfig()
        config.Net.DialTimeout = 10 * time.Second
        config.Net.SASL.Enable = true
        config.Net.SASL.User = "$ConnectionString"
        config.Net.SASL.Password = getEnv(eventHubsConnStringEnvVar)
        config.Net.SASL.Mechanism = "PLAIN"

        config.Net.TLS.Enable = true
        config.Net.TLS.Config = &tls.Config{
            InsecureSkipVerify: true,
            ClientAuth:         0,
        }
        config.Version = sarama.V1_0_0_0

        return config
    }

Authentication is the key here - Azure Event Hubs (including the Kafka features) requires SSL or TLS for all communication and uses Shared Access Signatures (SAS) for authentication. 

> Please refer to [the documentation](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-authentication-and-security-model-overview?WT.mc_id=devto-blog-abhishgu) for details regarding the security model

- SASL is enabled with `config.Net.SASL.Enable = true` and uses the `PLAIN` mechanism (config.Net.SASL.Mechanism)
- SASL User is defined using `config.Net.SASL.User` and is set to `$ConnectionString` (yes this is static)
- SASL password (`config.Net.SASL.Password`) is set to the Event Hubs connection string (details to follow in the upcoming section)
- Finally, TLS is enabled (`config.Net.TLS.Enable`) and `config.Net.TLS.Config` is set to an instance of [TLS Config](https://godoc.org/crypto/tls#Config)

> The configuration for the producer and consumer (for this example) is the same except for `config.Producer.Return.Successes = true` which is required for the synchronous version of the producer

Things to watch out for:

- If you do not configure TLS properly, you might see this error - `Failed to start Sarama producer: kafka: client has run out of available brokers to talk to (Is your cluster reachable?)`
- At the time of writing this article, [Event Hubs does not support message compression](https://docs.microsoft.com/azure/event-hubs/event-hubs-for-kafka-ecosystem-overview?WT.mc_id=devto-blog-abhishgu#features-that-are-not-yet-supported), so please ensure that you use the default value of `config.Producer.Compression` (which is `none`) or set it explicitly using `config.Producer.Compression = sarama.CompressionNone`. If not, you might see this error `Failed to send msg: kafka server: The requested operation is not supported by the message format version.`
- Make sure you set the Kafka broker version to `config.Version = sarama.V1_0_0_0` or higher

#### Event Hubs Kafka Producer

Create a producer instance passing in the Event Hubs broker and the required configuration (`sarama.Config`)

	producer, err := sarama.NewSyncProducer(brokerList, getConfig())
	if err != nil {
		fmt.Println("Failed to start Sarama producer:", err)
		os.Exit(1)
	}

Start a goroutine for producing messages. This happens in an infinite for loop which can be stopped using `ctrl+c`

	go func() {
		for {
			if producerOpen {
				ts := time.Now().String()
				msg := &sarama.ProducerMessage{Topic: eventHubsTopic, Key: sarama.StringEncoder("key-" + ts), Value: sarama.StringEncoder("value-" + ts)}
				p, o, err := producer.SendMessage(msg)
				if err != nil {
					fmt.Println("Failed to send msg:", err)
					continue
				}
			}
			...
		}
	}()

For a clean exit, listen to the interrupt signal (`ctrl+c`) and close the producer

	close := make(chan os.Signal)
	signal.Notify(close, syscall.SIGTERM, syscall.SIGINT)
	<-close
	err = producer.Close()

#### Event Hubs Kafka Consumer

Create a consumer group instance passing in the Event Hubs broker

	consumer, err := sarama.NewConsumerGroup(brokerList, consumerGroupID, getConfig())

Start consuming in a separate goroutine

	go func() {
		for {
			err = consumer.Consume(ctx, []string{getEnv(eventHubsTopicEnvVar)}, messageHandler{})
			....
			if ctx.Err() != nil {
				return
			}
		}
	}()


`messageHandler` implements [`sarama.ConsumerGroupHandler`](https://godoc.org/github.com/Shopify/sarama#ConsumerGroupHandler) (functions - `Setup`, `Cleanup` and `ConsumeClaim`). `ConsumeClaim` function is the important one. It is where you specify what to do with each message - in this example, it is logged to standard out and marked as consumed

    func (h messageHandler) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error {
        for msg := range c.Messages() {
            fmt.Printf("Message topic:%q partition:%d offset:%d\n", msg.Topic, msg.Partition, msg.Offset)
            fmt.Println("Message content", string(msg.Value))
            s.MarkMessage(msg, "")
        }
        return nil
    }


The consumer can be stopped by exiting the program (pressing `ctrl+c`). For a clean exit, the consumer instance is closed

	close := make(chan os.Signal)
	signal.Notify(close, syscall.SIGTERM, syscall.SIGINT)
	<-close
	cancel()
	if err := consumer.Close(); err != nil {
        ......
	}

Ok, let's create an Event Hubs cluster and try the end to end scenario.

### Create your Kafka enabled Event Hubs cluster

If you have a cluster already, skip this and go to the *"Event Hubs connection details"* sub-section

Set environment variables:

    export AZURE_SUBSCRIPTION=[to be filled]
    export AZURE_RESOURCE_GROUP=[to be filled]
    export AZURE_LOCATION=[to be filled]
    export EVENT_HUBS_NAMESPACE=[name of the event hub namespace - to be filled]
    export EVENT_HUB_NAME=[name of the event hub (topic) - to be filled]

Create the resource group if you don't have one already

    az account set --subscription $AZURE_SUBSCRIPTION
    az group create --name $AZURE_RESOURCE_GROUP --location $AZURE_LOCATION

Create an Event Hubs namespace (similar to a Kafka Cluster)

> For details on Event Hubs namespace, please refer to the [Event Hubs documentation](https://docs.microsoft.com/azure/event-hubs/event-hubs-features?WT.mc_id=devto-blog-abhishgu)

    az eventhubs namespace create --name $EVENT_HUBS_NAMESPACE --resource-group $AZURE_RESOURCE_GROUP --location $AZURE_LOCATION --enable-kafka true --enable-auto-inflate false

> Documentation for [`az eventhubs namespace create`](https://docs.microsoft.com/cli/azure/eventhubs/namespace?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-eventhubs-namespace-create)

And then create an Event Hub (same as a Kafka topic)

    az eventhubs eventhub create --name $EVENT_HUB_NAME --resource-group $AZURE_RESOURCE_GROUP --namespace-name $EVENT_HUBS_NAMESPACE --partition-count 10

> Documentation for [`az eventhub create`](https://docs.microsoft.com/cli/azure/eventhubs/eventhub?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-eventhubs-eventhub-create)


#### Event Hubs connection details

Get the connection string and credentials for your cluster

> For details, read [how Event Hubs uses Shared Access Signatures for authorization](https://docs.microsoft.com/azure/event-hubs/authorize-access-shared-access-signature?WT.mc_id=devto-blog-abhishgu)

Start by fetching the Event Hub rule/policy name

    az eventhubs namespace authorization-rule list --resource-group $AZURE_RESOURCE_GROUP --namespace-name $EVENT_HUBS_NAMESPACE

> Documentation for [`az eventhubs namespace authorization-rule list`](https://docs.microsoft.com/cli/azure/eventhubs/namespace/authorization-rule?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-eventhubs-namespace-authorization-rule-list)


You will get a JSON output similar to below:

    [
        {
            "id": "/subscriptions/qwerty42-ae29-4924-b6a7-dda0ea91d347/resourceGroups/foobar-resource/providers/Microsoft.EventHub/namespaces/foobar-event-hub-ns/AuthorizationRules/RootManageSharedAccessKey",
            "location": "Southeast Asia",
            "name": "RootManageSharedAccessKey",
            "resourceGroup": "foobar-resource",
            "rights": [
            "Listen",
            "Manage",
            "Send"
            ],
            "type": "Microsoft.EventHub/Namespaces/AuthorizationRules"
        }
    ]

The authorization rule name is the value of the `name` attribute (without the quotes), which in this case is `RootManageSharedAccessKey`

    export EVENT_HUB_AUTH_RULE_NAME=RootManageSharedAccessKey
    
And, then make use of the rule name to extract the connection string

    az eventhubs namespace authorization-rule keys list --resource-group $AZURE_RESOURCE_GROUP --namespace-name $EVENT_HUBS_NAMESPACE --name $EVENT_HUB_AUTH_RULE_NAME

> Documentation for [`az eventhubs namespace authorization-rule keys list`](https://docs.microsoft.com/cli/azure/eventhubs/namespace/authorization-rule/keys?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-eventhubs-namespace-authorization-rule-keys-list)


You'll get a JSON response as such:

    {
        "aliasPrimaryConnectionString": null,
        "aliasSecondaryConnectionString": null,
        "keyName": "RootManageSharedAccessKey",
        "primaryConnectionString": "Endpoint=sb://foobar-eventhub-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Nbaz0D42MT7qwerty6D/W51ao42r6EJuxR/zEqwerty=",
        "primaryKey": "qwertyEiQHIirSNDPzqcqvZEUs6VAW+JIK3L46tqwerty",
        "secondaryConnectionString": "Endpoint=sb://abhishgu-temp-event-hub-ns.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=qwertyPF2/YRGzxKmb06Z8NBFLCjnX38O7ch6aiYkN0=",
        "secondaryKey": "qwertyPF2/YRGzxKmb06Z8NBqwertyX38O7ch6aiYk42="
    }

The primary connection string is the value of the `primaryConnectionString` attribute (without the quotes), which in this case is `"Endpoint=sb://foobar-eventhub-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Nbaz0D42MT7qwerty6D/W51ao42r6EJuxR/zEqwerty="`. Make a note of the connection string as you will be using it in the next step.

> [This information is sensitive - please excercise caution](https://docs.microsoft.com/azure/event-hubs/authorize-access-shared-access-signature?&WT.mc_id=devto-blog-abhishgu#best-practices-when-using-sas)


### Test producer and consumer

Clone the GitHub repository and navigate to the right directory:

    git clone https://github.com/abhirockzz/eventhubs-kafka-go-sarama
    cd eventhubs-kafka-go-sarama
    
Fetch the Sarama Kafka client library

    go get github.com/Shopify/sarama


#### Producer

Set environment variables

    export EVENTHUBS_CONNECTION_STRING=[value of primary connection string obtained in the previous step]
    export EVENT_HUBS_NAMESPACE=[event hub namespace]
    export EVENTHUBS_BROKER=$EVENT_HUBS_NAMESPACE.servicebus.windows.net:9093
    export EVENTHUBS_TOPIC=[name of the event hub (topic)]

> for `EVENTHUBS_CONNECTION_STRING` variable, please ensure that you *include* the double-quotes in the value received using the Azure CLI e.g. `export EVENTHUBS_CONNECTION_STRING="Endpoint=sb://foobar-eventhub-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Nbaz0D42MT7qwerty6D/W51ao42r6EJuxR/zEqwerty="`

Start the producer

    go run producer/sarama-producer.go


Once it starts, you should see the logs

    Event Hubs broker [foo-bar.servicebus.windows.net:9093]
    Event Hubs topic testhub
    Waiting for program to exit...
    sent message to partition 0 offset 1
    sent message to partition 7 offset 1
    sent message to partition 6 offset 1
    sent message to partition 8 offset 1
    sent message to partition 2 offset 1

> To stop, just press ctrl+c on your terminal

#### Consumer

Start the consumer process in a different terminal. Set environment variables


    export EVENTHUBS_CONNECTION_STRING=[value of primary connection string obtained in the previous step]
    export EVENT_HUBS_NAMESPACE=[event hub namespace]
    export EVENTHUBS_BROKER=$EVENT_HUBS_NAMESPACE.servicebus.windows.net:9093
    export EVENTHUBS_TOPIC=[name of the event hub (topic) - to be filled]
    export EVENTHUBS_CONSUMER_GROUPID=[name of consumer group e.g. testgroup]


> for `EVENTHUBS_CONNECTION_STRING` variable, please ensure that you *include* the double-quotes in the value received using the Azure CLI e.g. `export EVENTHUBS_CONNECTION_STRING="Endpoint=sb://foobar-eventhub-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Nbaz0D42MT7qwerty6D/W51ao42r6EJuxR/zEqwerty="`

Start the consumer

    go run consumer/sarama-consumer.go

In the logs, you will see that the consumer group gets created and all the partitions (10 in this example) are allocated to it

    Event Hubs broker [foo-bar.servicebus.windows.net:9093]
    Sarama client consumer group ID testgroup
    new consumer group created
    Event Hubs topic testhub
    Waiting for program to exit
    Partition allocation - map[testhub:[0 1 2 3 4 5 6 7 8 9]]
    Message topic:"testhub" partition:9 offset:45
    Message content value-2019-10-08 16:12:23.704802 +0530 IST m=+1.003667284
    Message topic:"testhub" partition:3 offset:32
    Message content value-2019-10-08 17:05:42.388301 +0530 IST m=+0.912420074

#### Scale out...

In a different terminal, start another instance of the consumer. This will trigger a rebalance of the partitions and you will see that few (5 in this case) will get allocated to this (new) consumer instance

    Event Hubs broker [foo-bar.servicebus.windows.net:9093]
    Sarama client consumer group ID testgroup
    new consumer group created
    Event Hubs topic testhub
    Waiting for program to exit
    Partition allocation - map[testhub:[0 1 2 3 4]]

If you go back to the terminal for the first consumer instance, you will see that few partitions have been taken away as a result of the rebalancing

    Consumer group clean up initiated
    Partition allocation - map[testhub:[5 6 7 8 9]]

> To stop, just press `ctrl+c` on your terminal

Now, both the consumers will share the workload and consume messages from Event Hubs. You keep scaling out by starting more consumer instances, but this will only be useful until you reach the point where the number of consumer instances is equal to the number of partitions. In essence, the number of partitions of your Events Hub is the unit of parallelism and scale.

That's all for this blog. I would love to have your feedback and suggestions! Don't be shy, just [tweet](https://twitter.com/abhi_tweeter) or drop a comment. And, if you found this article useful, please like and follow 😃😃

![](https://media.giphy.com/media/26FPvntWVkwaRWgXS/giphy-downsized.gif)