Digital Twin MQTT-Kafka Kubernetes

Helm Commands

Infra

• helm install infra-fpknebel infra/ --namespace=fpknebel
• helm uninstall infra-fpknebel --namespace=fpknebel
• helm upgrade --install infra-fpknebel infra/ --namespace=fpknebel
• helm uninstall infra-fpknebel --namespace=fpknebel --debug && helm install infra-fpknebel infra/ --namespace=fpknebel --debug

Twin

• helm install twin-fpknebel twin/ --namespace=fpknebel
• helm uninstall twin-fpknebel --namespace=fpknebel
• helm upgrade --install twin-fpknebel twin/ --namespace=fpknebel
• helm uninstall twin-fpknebel --namespace=fpknebel --debug && helm install twin-fpknebel twin/ --namespace=fpknebel --debug

Pod Commands

Execute command in container inside pod

kubectl exec kafka-0 -c kafka-connect --namespace fpknebel -- [COMMAND]

Create sink

kubectl exec kafka-0 -c kafka-connect --namespace fpknebel -- \
curl -X POST http://localhost:8083/connectors \
-H 'Content-Type: application/json' \
-d '
{ 
    "name": "mqtt-sink",
    "config": {
        "connector.class": "io.confluent.connect.mqtt.MqttSinkConnector",
        "tasks.max": 1,
        "mqtt.server.uri": "tcp://mqtt-cluster:1883",
        "topics": "mqtt-sink",
        "mqtt.qos": "1",
        "confluent.topic.bootstrap.servers": "LISTENER_EXTERNAL://kafka:29092",
        "confluent.topic.replication.factor": 1,
        "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
        "value.converter.schemas.enable": false
    }
}'

Create source

kubectl exec kafka-0 -c kafka-connect --namespace fpknebel -- \
curl -X POST http://localhost:8083/connectors \
-H 'Content-Type: application/json' \
-d '
{
    "name": "mqtt-source",
    "config": {
        "connector.class": "io.confluent.connect.mqtt.MqttSourceConnector",
        "tasks.max": "1",
        "mqtt.server.uri": "tcp://mqtt-cluster:1883",
        "mqtt.topics": "topic/*",
        "kafka.topic": "mqtt-source",
        "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
        "confluent.topic.bootstrap.servers": "LISTENER_EXTERNAL://kafka:29092",
        "confluent.topic.replication.factor": 1
    }
}'

Get existing connectors

kubectl exec kafka-0 -c kafka-connect --namespace fpknebel -- \
curl -X GET http://localhost:8083/connectors

Kafka Commands

For testing directly inside the pods, if Kafka topics are producing/consuming.

Producer

kafka-console-producer --bootstrap-server LISTENER_EXTERNAL://kafka:29092 --topic mqtt-source --property parse.key=true --property key.separator=,

key,value

Consumer

kafka-console-consumer --bootstrap-server LISTENER_EXTERNAL://kafka:29092 --topic mqtt-source

License

MIT License. Read more about it here.