Set up Kafka extension

Prequisite: A running HiveMQ Cluster Install HiveMQ on the AKS cluster

If you are not logged in, please use the following commands to log in to your azure cluster, please replace the group and name as needed. (Our reference : https://hivemq.atlassian.net/wiki/spaces/HMS/pages/2691203114/Setting+up+AKS+Cluster+in+Azure#Set-Up-Your-Kubernetes-Cluster-With-AKS)

az login

az aks get-credentials -g hmqResourceGroup -n HiveMQCluster

Install Kafka using helm

Create a namespace for Kafka and switch the context to it:

kubectl create namespace kafka; 
kubectl config set-context --current --namespace=kafka

Add the repository for the Kafka Helm chart to your package manager.

helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update

Deploy the Kafka server using the Helm chart. The below command deploys Kafka with 2 brokers (replicas).
```
helm upgrade --install kafka bitnami/kafka --namespace=kafka --set replicaCount=2 
```
1. If everything is correct, then
  1. Kafka can be accessed by consumers via port 9092 on the following DNS name from within your cluster:
    kafka.kafka.svc.cluster.local
  2. The CLIENT listener for Kafka client connections from within your cluster has been configured with the following security settings:
    - SASL authentication
  3. To connect a client to your Kafka:
    1. username="user1"
    2. To get the password execute the command below:(skip % at the end)
      kubectl get secret kafka-user-passwords --namespace kafka \ -o jsonpath='{.data.client-passwords}' | base64 -d | cut -d , -f 1;

Enabling Kafka extension, Configuring the extension and Kafka

HiveMQ Enterprise Extension For Kafka requires a separate license file, e.g. kafka-license.elic, in the $HIVEMQ_HOME/license directory. You can skip this step. If you skip this step, then the kafka-extension will start in trial mode, limited to 5h, and will be automatically disabled by the HiveMQ broker after 5h. To add the kafka-license.elic along with the hivemq-license.lic, create a new configmap hivemq-license including all desired license files:
```
kubectl create configmap hivemq-license --namespace=hivemq \
  --from-file hivemq-licesen.lic \
  --from-file kafka-license.elic
```

Edit the values.yaml file of the hivemq-operator, section hivemq.configMaps. Update this:

  configMaps: []
  # ConfigMaps to mount to the HiveMQ pods. These can be mounted to existing directories without shadowing the folder contents as well.
  #- name: hivemq-license
  #  path: /opt/hivemq/license

To this:

  configMaps: 
    - name: hivemq-license
      path: /opt/hivemq/license

This will mount the content of the configMap hivemq-license to the directory /opt/hivemq/license of the hivemq-broker pods.

HiveMQ Enterprise Extension For Kafka is preinstalled with HiveMQ so once you enable it, it will look for its configuration file. You must prepare this file before enabling the extension. If you skip this step, the extension will not find its configuration file and will not load any configuration.
Prepare a simple configuration file for kafka-extension as in the example below.
- this example configuration will map all incoming MQTT publish packets to the topic “test” in Kafka; and will map the topic “test” in Kafka to the topic “test-test” in the HiveMQ broker
- Use your password in <password>here_is_your_password</password>, that you successfully retrieved with this command a few steps ago:
```
kubectl get secret kafka-user-passwords --namespace kafka \
  -o jsonpath='{.data.client-passwords}' | base64 -d | cut -d , -f 1;
```
- Here is the file:

config.xml

<?xml version="1.0" encoding="UTF-8" ?>
<kafka-configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                     xsi:noNamespaceSchemaLocation="config.xsd">
    <kafka-clusters>
        <kafka-cluster>
            <id>cluster01</id>
            <bootstrap-servers>kafka.kafka.svc.cluster.local:9092</bootstrap-servers>
            <authentication>
                <scram-sha256>
                    <username>user1</username>
                    <password>here_is_your_password</password>
                </scram-sha256>
            </authentication>
        </kafka-cluster>
    </kafka-clusters>

    <mqtt-to-kafka-mappings>
        <mqtt-to-kafka-mapping>
            <id>mapping01</id>
            <cluster-id>cluster01</cluster-id>
            <mqtt-topic-filters>
                <mqtt-topic-filter>#</mqtt-topic-filter>
            </mqtt-topic-filters>
            <kafka-topic>test</kafka-topic>
        </mqtt-to-kafka-mapping>
    </mqtt-to-kafka-mappings>

    <kafka-to-mqtt-mappings>
        <kafka-to-mqtt-mapping>
            <id>mapping02</id>
            <cluster-id>cluster01</cluster-id>
            <kafka-topics>
                <kafka-topic>test</kafka-topic>
                <kafka-topic-pattern>test-(.)*</kafka-topic-pattern>
            </kafka-topics>
        </kafka-to-mqtt-mapping>
    </kafka-to-mqtt-mappings>
</kafka-configuration>

Create a new configMap kafka-config from the config.xml:

kubectl create configmap kafka-config --from-file config.xml

Edit values.yaml file of hivemq-operator and update section hivemq.extensions, having name: hivemq-kafka-extension
Update this:

  extensions:
    - name: hivemq-kafka-extension
      extensionUri: preinstalled
      enabled: false

To this:

extensions:
    - name: hivemq-kafka-extension
      extensionUri: preinstalled
      enabled: true
      configMap: kafka-config
      initialization: |
        # Fixes the location of the config.xml file
        [[ ! -f conf/config.xml ]] &&
        [[ -f /conf-override/extensions/hivemq-kafka-extension/config.xml ]] &&
        ln -s /conf-override/extensions/hivemq-kafka-extension/config.xml conf/config.xml

Re-deploy hivemq-operator with updated values.yaml

helm upgrade hivemq --install hivemq/hivemq-operator --values my-hivemq-values.yaml --namespace hivemq

If everything is correct,

The HiveMQ log contains info about using the correct license:

kubectl logs deployment/hivemq | grep 'Using valid'

INFO  - Using valid Enterprise Edition CPU license (hivemq-license.lic) issued to HiveMQ - Internal for max 9999 CPU cores, valid until 2024-03-31.
INFO  - Using valid license (kafka-license.elic) for enterprise extension with name "HiveMQ Enterprise Extension for Kafka", valid until 2024-03-31.

The HiveMQ log contains info about successful Kafka connection:

kubectl logs deployments/hivemq -f | grep -i kafka

INFO  - Starting extension with id "hivemq-kafka-extension" at /opt/hivemq/extensions/hivemq-kafka-extension
INFO  - No mqtt-to-kafka transformers were found. No mqtt-to-kafka transformers are started.
INFO  - Started mqtt-to-kafka mapping "mapping01" with MQTT topics [#] and Kafka topic "test" for Kafka Cluster "cluster01"
INFO  - No kafka-to-mqtt transformers were found. No kafka kafka-to-mqtt transformers are started.
INFO  - Started kafka-to-mqtt mapping "mapping02" with kafka topic: [test] and kafka pattern: [test-(.)*] for Kafka Cluster "cluster01" (poll duration: 100 ms).
INFO  - Extension "HiveMQ Enterprise Extension for Kafka" version 4.24.0 started successfully.
INFO  - Subscribing kafka-to-mqtt topic-mapping with id mapping02 to kafka pattern (test-(.)*|\Qtest\E)

Kafka dashboard is visible in the HiveMQ Control Center:

End-to-end testing

Subscribe a reference mqtt client to the topic “test”, output show the topic name and message:
```
mqtt subscribe --topic '#' --host $hivemqhost --port 1883 -q 1 --showTopics
```
Do not close this terminal session!

From a different terminal session, publish a message to the topic “test”:

mqtt publish --topic test --message Hello --host $hivemqhost --port 1883 -q 1

If everything is correct, the subscriber will indefinitely receive the message we published:
```
mqtt subscribe --topic '#' --host $hivemqhost --port 1883 -q 1 --showTopics

test: Hello
test: Hello
test: Hello
test: Hello
test: Hello
test: Hello
test: Hello
test: Hello
test: Hello
```
You can stop this by terminating the execution of the command by pressing Ctrl+C.
To gain extra points for this training, explain, why the message is received indefinitely rather than once.

If everything is correct, the Kafka Dashboard in the HiveMQ Control Center shows incoming and outgoing Kafka messaging: