update tests and adjust a couple env var things

Author: tim-quix

python/destinations/MQTT/library.json

@@ -80,15 +80,15 @@
       "Required": true
     },
     {
-      "Name": "mqtt_tls_enabled",
+      "Name": "MQTT_RETAIN_MESSAGES",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
-      "Description": "Enable TLS for MQTT connection (true/false)",
-      "DefaultValue": "true",
+      "Description": "Whether to retain/persist the latest message (per key) on the MQTT topic.",
+      "DefaultValue": "false",
       "Required": false
     },
     {
-      "Name": "consumer_group_name",
+      "Name": "CONSUMER_GROUP",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
       "Description": "Kafka consumer group name for the MQTT destination",

python/destinations/MQTT/main.py

@@ -1,13 +1,16 @@
-from mqtt import MQTTSink
 from quixstreams import Application
+from quixstreams.sinks.community.mqtt import MQTTSink
 import os
 
 # Load environment variables (useful when working locally)
 # from dotenv import load_dotenv
 # load_dotenv()
 
-app = Application(consumer_group="mqtt_consumer_group", auto_offset_reset="earliest")
-input_topic = app.topic(os.environ["input"], value_deserializer="double")
+app = Application(
+    consumer_group=os.getenv("CONSUMER_GROUP", "mqtt_consumer_group"),
+    auto_offset_reset="earliest"
+)
+input_topic = app.topic(os.environ["input"])
 
 sink = MQTTSink(
     client_id=os.environ["MQTT_CLIENT_ID"],
@@ -17,6 +20,7 @@
     username=os.environ["MQTT_USERNAME"],
     password=os.environ["MQTT_PASSWORD"],
     version=os.environ["MQTT_VERSION"],
+    retain=os.getenv("MQTT_RETAIN_MESSAGES", "false").lower() == "true",
     tls_enabled=os.environ["MQTT_USE_TLS"].lower() == "true"
 )
 

python/destinations/MQTT/mqtt.py

@@ -1,2 +1,2 @@
-quixstreams==3.23.1
+quixstreams[mqtt]==3.23.1
 python-dotenv

python/destinations/MQTT/requirements.txt

@@ -2,12 +2,17 @@
 services:
   mqtt-broker:
     image: eclipse-mosquitto:latest
+    environment:
+      - MQTT_USERNAME=testuser
+      - MQTT_PASSWORD=testpass
+    entrypoint: ["/bin/sh", "/init-mqtt.sh"]
     networks:
       - test-network
     volumes:
       - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
+      - ./init-mqtt.sh:/init-mqtt.sh:ro
     healthcheck:
-      test: ["CMD-SHELL", "mosquitto_sub -t '$$SYS/#' -C 1 -i healthcheck -W 3 || exit 1"]
+      test: ["CMD-SHELL", "mosquitto_sub -t '$$SYS/#' -C 1 -i healthcheck -W 3 -u testuser -P testpass || exit 1"]
       interval: 3s
       timeout: 5s
       retries: 10
@@ -35,13 +40,17 @@ services:
       dockerfile: Dockerfile
     environment:
       - Quix__Broker__Address=kafka:9092
-      - consumer_group_name=mqtt-test-consumer
       - input=test-mqtt-input
-      - mqtt_server=mqtt-broker
-      - mqtt_port=1883
-      - mqtt_topic_root=test/output
-      - mqtt_version=3.1.1
-      - mqtt_tls_enabled=false
+      - CONSUMER_GROUP=mqtt-test-consumer
+      - MQTT_CLIENT_ID=test_client
+      - MQTT_SERVER=mqtt-broker
+      - MQTT_PORT=1883
+      - MQTT_TOPIC_ROOT=test/output
+      - MQTT_VERSION=3.1.1
+      - MQTT_USE_TLS=false
+      - MQTT_USERNAME=testuser
+      - MQTT_PASSWORD=testpass
+      - MQTT_RETAIN_MESSAGES=true
     networks:
       - test-network
     depends_on:
@@ -61,6 +70,8 @@ services:
       - MQTT_BROKER=mqtt-broker
       - MQTT_PORT=1883
       - MQTT_TOPIC=test/output/#
+      - MQTT_USERNAME=testuser
+      - MQTT_PASSWORD=testpass
     command: >
       sh -c "
         echo 'Installing MQTT client...' &&

tests/destinations/MQTT/docker-compose.test.yml

@@ -0,0 +1,21 @@
+#!/bin/sh
+# Initialize MQTT broker with authentication
+
+MQTT_USERNAME="${MQTT_USERNAME:-testuser}"
+MQTT_PASSWORD="${MQTT_PASSWORD:-testpass}"
+CONFIG_DIR="/app/state/mosquitto/config"
+LOG_DIR="/app/state/mosquitto/log"
+DATA_DIR="/app/state/mosquitto/data"
+
+echo "Setting up MQTT broker directories..."
+mkdir -p "$CONFIG_DIR" "$LOG_DIR" "$DATA_DIR"
+
+echo "Creating password file for user: $MQTT_USERNAME"
+mosquitto_passwd -b -c "$CONFIG_DIR/passwd" "$MQTT_USERNAME" "$MQTT_PASSWORD"
+
+echo "Setting permissions for mosquitto user..."
+chown -R mosquitto:mosquitto /app/state/mosquitto
+chmod -R 755 /app/state/mosquitto
+
+echo "Starting Mosquitto MQTT broker..."
+exec mosquitto -c /mosquitto/config/mosquitto.conf

tests/destinations/MQTT/init-mqtt.sh

@@ -1,2 +1,6 @@
 listener 1883
-allow_anonymous true
+allow_anonymous false
+persistence true
+persistence_location /app/state/mosquitto/data/
+log_dest file /app/state/mosquitto/log/mosquitto.log
+password_file /app/state/mosquitto/config/passwd