Publish to public DSH stream¶

This page describes the steps to publish messages to a Kafka topic that is part of a public DSH stream. You will create a very simple Kafka producer in Python that sends messages to a public DSH stream every five seconds.

Prerequisites¶

Before you can follow this tutorial, you need the following:

On the DSH:
- Access to a tenant on a DSH platform
- A public DSH stream. See Adding a DSH stream for more information:
  - Your tenant needs the “Read/Write” permissions for this DSH stream.
  - Use the default values for replication factor, number of partitions, retained messages, and topic level for partitioning.
  - You also need an API client that has “PUB” permission and “SUB” permission on the MQTT topics for this public DSH stream if you want to access it via MQTT.
- Access to the Harbor container image registry, and the username and CLI secret for your user. See Accessing Harbor for more information.
- The Grafana service. See Requesting Prometheus and Grafana for more information.
On your machine:
- A Unix-based system, for example Linux, MacOS or Windows Subsystem for Linux
- Docker CLI
- Protoc, the Protocol Buffer Compiler

Key concepts¶

If you want to interact with a public DSH stream, then you need to understand some key concepts. Take your time to read through the sections below, and to discover the details in the reference pages.

Public DSH stream¶

A public DSH stream is a collection of Kafka topics, and it will contain at least 2 of them. If a tenant ‘automator’ requested a public DSH stream ‘temperature’, then this DSH stream contains at least the following Kafka topics:

stream.temperature.automator: The Kafka topic that the tenant ‘automator’ publishes messages to with a Kafka producer.
stream.temperature.dsh: The Kafka topic that the DSH publishes messages to that come in via the Messaging API.

You can share a public DSH stream with multiple tenants, who then receive their own Kafka topic inside the DSH stream. However, only the DSH can write to a *.dsh Kafka topic.

Note

See DSH stream for more information.

Messaging API¶

You can expose a public DSH stream via the DSH’s Messaging API. If you do so, external clients can publish messages to the public DSH stream via MQTT or HTTP, and they can subscribe to the DSH stream:

The messages in a public DSH stream support the MQTT protocol. As a consequence, these messages have metadata specifically for MQTT.
The DSH publishes messages that come in via the Messaging API to a specific topic in a public DSH stream. In a DSH stream called ‘temperature’, the DSH writes these messages to the Kafka topic stream.temperature.dsh.
When an external HTTP client or MQTT client subscribes to an MQTT topic in a public DSH stream, it consumes the messages across all Kafka topics in that DSH stream, provided that these messages have a matching MQTT topic.

Note

See Messaging API for more information.

Message envelopes¶

In order to support the MQTT protocol, messages in a public DSH stream contain metadata for “MQTT topic”, “Quality of Service”, “retained”, and the identity of the publisher. For that reason, a Kafka producer must wrap the messages in Protobuf message envelopes:

The KeyEnvelope contains the metadata, and the DataEnvelope contains the message itself.
These message envelopes allow the Messaging API to handle these messages correctly.
The Kafka producer must also apply a specific partitioning scheme to ensure that messages with the same MQTT topic end up in the same Kafka partition. The Messaging API follows the same partitioning scheme. That way, you maintain the order and history of messages.

Note

See DSH stream and Message envelopes for more information.

Create your files¶

The steps below describe how you can create the files for your container image.

Working directory¶

Open the Terminal, create a directory for this tutorial, and enter it:

Terminal

mkdir public-dsh-stream-produce
cd public-dsh-stream-produce

Bash scripts¶

In the working directory, create two bash files:

set_up_config.sh: Set up the configuration to connect to the DSH’s Kafka cluster.
entrypoint.sh: Execute the set_up_config.sh script, and then execute the Python script in the current shell to ensure that termination signals are handled properly.

Bash script for configuration¶

This script configures the SSL connection with the Kafka cluster. It stores the keys and certificates in the /home/dsh/pki/ directory of your container, and stores information about the Kafka configuration as environment variables.

In the working directory, create a file set_up_config.sh, with the contents of the Script to configure SSL for Kafka.

Bash script for entrypoint¶

This script is the default executable for your service’s container. It executes the set_up_config.sh script, and then executes in the current shell any subsequent commands.

In the working directory, create a file called entrypoint.sh, with the contents below:

entrypoint.sh
#!/usr/bin/env bash

# Find the parent directory of the file.
medir=${0%/*}

# Execute the bash script for the SSL and Kafka configuration.
source ${medir}/set_up_config.sh

# Override the current shell session without creating a new process, and use the parameters passed to the script as commands. The Dockerfile passes the Python command as a parameter.
exec "$@"

Protobuf message envelopes¶

This file defines the message envelopes that Kafka clients must use to interact with public DSH streams. It defines the KeyEnvelope for the message metadata, and the DataEnvelope for the payload. In order to use these envelopes, you need to create a Python module from the Protobuf file:

In the working directory, create a file envelopes.proto, with the contents of the Protobuf file for message envelopes.
In the Terminal, execute the command below. It creates the Python module file envelopes_pb2.py that you will import in the nexts step.

Terminal

protoc -I=. --python_out=. ./envelopes.proto

Python script¶

This script does the actual work of producing messages to the tenant’s Kafka topic inside the public DSH stream:

It retrieves the configuration for the SSL connection and the Kafka cluster.
It creates a Producer object.
It calculates the partition, using the topic level depth and number of partions of the DSH stream, and the MQTT topic of the message.
It wraps the metadata and the value in the appropriate envelopes.
It publishes the message and creates a log entry to register the success or failure to publish messages.

main.py
import os
import sys
import json
import time
import signal
import uuid
import envelopes_pb2 as proto
from confluent_kafka import Producer
from kafka.partitioner import murmur2

# Retrieve the tenant configuration that set_up_config.sh created.
tenant_cfg = os.getenv("JSON_TENANT_CONFIG")

# Handle termination signals gracefully.
def handle_sigterm(signum, frame):
    print("Received SIGTERM, shutting down.")
    sys.exit(0)

# Set the configuration of the Kafka producer, using the tenant configuration that set_up_config.sh created.
def load_config():
    # Retrieve the certificates, keys, and the ID of the service from the environment variables.
    pki_cacert = os.environ["DSH_PKI_CACERT"]
    pki_key = os.environ["DSH_PKI_KEY"]
    pki_cert = os.environ["DSH_PKI_CERT"]
    client_id = os.environ["MESOS_TASK_ID"]

    # Retrieve the addresses of the Kafka brokers from the environment variable with the JSON configuration, or from the environment variable with the Kafka servers.
    if tenant_cfg:
        cfg = json.loads(tenant_cfg)
        servers = ",".join(cfg["brokers"])
    else:
        servers = os.environ.get("KAFKA_SERVERS")

    # Make sure that the servers are set, because the Producer object needs them.
    if not servers:
        print("Error: Kafka servers not set.")
        sys.exit(1)

    return {
        "bootstrap.servers": servers,
        "client.id": client_id,
        "security.protocol": "ssl",
        "ssl.key.location": pki_key,
        "ssl.certificate.location": pki_cert,
        "ssl.ca.location": pki_cacert,
    }

# Write the success or failure of message publication to standard error output, so we can monitor it in Grafana.
def delivery_callback(err, msg):
    if err:
        sys.stderr.write('Message failed delivery: %s\n' % err)
    else:
        sys.stderr.write('Message delivered to %s [%d] @ %d\n' % (msg.topic(), msg.partition(), msg.offset()))

# Calculate the partition for a given message, using the MQTT topic, and the DSH stream's topic level depth and number of partitions.
# (1) Reduce the MQTT topic to the number of levels defined in the DSH stream's topic level depth.
# (2) Hash the reduced MQTT topic, and add a bitmask. Then apply the modulo operation to it, with the DSH stream's number of partitions as divisor.
def dsh_partitioner(key, topic_depth, partition_count):
    key_depth = '/'.join( key.split('/')[:(topic_depth)])
    return (murmur2(key_depth.encode('utf8')) & 0x7fffffff) % partition_count

# Wrap the metadata and value in Protobuf envelopes
def wrap(tenant, service_id, retained, qos, key, msg):
    # Create a Protobuf Identity object, and assign the correct values to its fields.
    identity = proto.Identity()
    identity.tenant = tenant
    identity.application = service_id

    # Create a Protobuf KeyHeader object, and assign the correct values to its fields.
    key_header = proto.KeyHeader()
    key_header.identifier.CopyFrom(identity)
    key_header.retained = retained
    if qos == 0:
        key_header.qos = proto.QoS.BEST_EFFORT
    elif qos == 1:
        key_header.qos = proto.QoS.RELIABLE

    # Create a Protobuf KeyEnvelope object, and assign the correct values to its fields. Serialize the key to a string.
    key_envelope = proto.KeyEnvelope()
    key_envelope.header.CopyFrom(key_header)
    key_envelope.key = key
    serialized_key = key_envelope.SerializeToString()

    # Create a Protobuf DataEnvelope object, and assign the message, in binary format. Serialize the data to a string.
    data_envelope = proto.DataEnvelope()
    data_envelope.payload = msg.encode('utf8')
    serialized_data = data_envelope.SerializeToString()

    return (serialized_key, serialized_data)

def produce_messages():
    # Listen for termination signals and load the SSL and Kafka configuration.
    signal.signal(signal.SIGTERM, handle_sigterm)
    config = load_config()

    # Create the Producer object. From the environment variables, retrieve the name of the Kafka topic in the DSH stream, and the MQTT topic.
    producer = Producer(config)
    stream_topic = os.environ.get("PRODUCE_STREAM_TOPIC")
    mqtt_topic = os.environ.get("MQTT_TOPIC")

    # Check whether the stream topic and MQTT topic are defined correctly.
    if not stream_topic or not mqtt_topic:
        print("Error: PRODUCE_STREAM_TOPIC or MQTT_TOPIC not set as environment variables.")
        sys.exit(1)

    # Retrieve the topic level depth and the partition count of the DSH stream from the tenant configuration, and calculate the Kafka partition.
    if tenant_cfg:
        cfg = json.loads(tenant_cfg)
        stream_name = stream_topic.rsplit('.',1)[0]
        topic_depth = int(cfg["streams"][stream_name]["partitioningDepth"])
        partition_count = int(cfg["streams"][stream_name]["partitions"])
        dsh_partition =  dsh_partitioner(mqtt_topic, topic_depth, partition_count)
    else:
        sys.stderr.write("Can't calculate partition because the tenant configuration is missing\n")
        sys.exit(1)

    count = 0

    # Create a random message, wrap it in message envelopes, and send it every 5 seconds.
    while True:
        # Create a message with a random value.
        some_string = uuid.uuid4().hex
        value = f"message-{count}-{some_string}"

        envelopes = wrap(os.environ.get("TENANT_NAME"), os.environ.get("DSH_CONTAINER_DNS_NAME"), True, 0, mqtt_topic, value)

        # Produce the message to the correct partition in the Kafka topic of the DSH stream. The value is the serialized data envelope and the key is the serialized key envelope.
        try:
            producer.produce(topic=stream_topic, key=envelopes[0], value=envelopes[1], callback=delivery_callback, partition=dsh_partition)
            producer.flush()
        except Exception as err:
            sys.stderr.write('Failed to produce message: %s' % err)

        count += 1
        time.sleep(5)


if __name__ == "__main__":
    produce_messages()

Some aspects of this script are worth noting:

It’s recommended to make your container responsive to termination signals (line 14–17 and line 93).
The script logs results to standard error output for the sake of demonstration (line 49–53). However, this isn’t a recommended way of working if you deploy custom services to a production environment because this is too chatty.
The partitioner uses the MQTT topic, topic level depth and number of partitions to calculate the partition that the producer should write the message to (line 58–60):
- It first reduces the MQTT topic to the number of levels defined by the topic level depth. Note that the MQTT topics in question don’t have a leading forward slash (/).
- It then calculates the hash of the reduced MQTT topic, using the murmur2 function from the standard Kafka library. You have to add a bitmask & 0x7fffffff to account for the difference between Java and Python. The DSH uses the Java implementation of the murmur2 function to calculate the partition, and Java has no unsigned integer types. This script uses the Python implementation of the murmur2 function, but Python does support unsigned integer types. The bitmask makes sure that there is no difference in the hashes between Java and Python.
- It then applies the the modulo operation to the hash, using the number of partitions as a divisor.
- Make sure that you use this logic when you calculate a partition, and don’t forget to add the bitmask in Python. If the message ends up in the wrong partition, MQTT clients can’t access it.
If a Kafka producer publishes messages to a public DSH stream, then it must wrap them in message envelopes. The script imports the Python module that we created in the previous step (line 7) and populates the fields in the wrap() function (line 63–89, called on line 125).
The script uses the PRODUCE_STREAM_TOPIC and MQTT_TOPIC environment variables that you define in the service definition. See Deploy the custom service below for more information.

Dockerfile¶

This file contains the instructions and commands to assemble an image using Docker.

In the working directory, create a file Dockerfile with the contents below:

Dockerfile
# Use the Python official Docker image .
FROM python:3.13

# Create an environment variable for the tenant's user ID.
ENV id=<tenant-user-ID>

# Install the necessary packages for SSL and Kafka.
RUN apt-get update && apt-get install -y --no-install-recommends \
    openssl \
    curl \
    librdkafka-dev

# Create "appuser" user.
RUN useradd --uid $id appuser

# Copy the bash scripts and the Python script.
COPY --chown=$id:$id --chmod=0755 entrypoint.sh /home/dsh/app/
COPY --chown=$id:$id --chmod=0755 set_up_config.sh /home/dsh/app/
COPY --chown=$id:$id --chmod=0755 main.py /home/dsh/app/

# Install the Confluent-Kafka Python package.
WORKDIR /home/dsh/app/
RUN pip install confluent-kafka google protobuf kafka_python

# Switch from root to appuser, execute the entrypoint bash script, and execute the Python script.
USER $id=$id
ENTRYPOINT ["/bin/bash", "/home/dsh/app/entrypoint.sh"] 
CMD ["python", "/home/dsh/app/main.py"]

Some aspects of this script are worth noting:

It’s important that you don’t run commands in your image as the root user:
- Running commands as root raises security issues.
- As a solution, you can add a new user and set it as the default user to run commands (line 14 and 26).
It’s recommended that you assign your tenant’s user ID to the new user:
- In the menu bar of the DSH Console, navigate to “Resources” > “Overview” to see the user ID of your tenant.
- You can add your tenant’s user ID as an environment variable (line 5).
- Specify the user ID via the environment variable when you create the user (line 14), and when you switch to the user (line 26).
The container executes the entrypoint.sh script (line 27) by default:
- As described in Bash script for configuration, this script executes the configuration script, and then uses the parameters passed to the script as commands.
- The CMD instruction passes the python command to entrypoint.sh, with the location of your Python script as a parameter (line 28).
- This is a standard way to set environment variables, and to make sure that all processes in the container are responsive to termination signals.
The Python script uses the following Python packages:
- confluent-kafka for the Kafka producer
- google and protobuf for the message envelopes
- kafka_python for the murmur2 hashing function

Build the image¶

Log in to the DSH container image registry¶

In the next step, log in to the Harbor container image registry of the DSH. Execute the command below, and enter the CLI secret for your user when prompted:

Terminal

docker login registry.cp.kpn-dsh.com -u <your-Harbor-username>

Replace <your-Harbor-username> with your actual Harbor username.
See Accessing Harbor for more information about Harbor and the credentials.

Build and push the container image¶

Now that you have access, you can actually build the container image using Docker, and push it to the DSH’s container image registry:

Terminal

docker build -t registry.cp.kpn-dsh.com/<your-tenant-name>/python-publish-stream:1.0.0 .
docker push registry.cp.kpn-dsh.com/<your-tenant-name>/python-publish-stream:1.0.0

Replace <your-tenant-name> with the name of your tenant on the DSH.
It’s recommended to tag your container images. For that reason, the code snippet uses the -t (or --tag) option, and the image has the name:tag format.
It’s recommended to use semantic versioning for the tag, which applies the pattern <major>.<minor>.<patch> for version numbers.
You’re free to choose a different name for your image.

Deploy the custom service¶

Finally, you need to add a custom service, and set up the service definition:

Click “Services” > “Overview” in the menu bar of the DSH Console.
Click the “+ Service” button at the top of the “Services” overview page.
Enter the name for the service, for example ‘python-publish-to-stream’.
Edit the JSON file for the service definition so that it has the form in the code snippet below. Don’t forget to replace the variables with the correct values:
- <your-tenant-name>: Your tenant’s name
- <tenant-user-ID>: Your tenant’s user ID. You can find it in the DSH Console, on the “Resources” overview page.
- <tenant-topic-in-public-stream>: The name of your tenant’s Kafka topic in the public DSH stream that you created for this tutorial. It has the format stream.<public-DSH-stream-name>.<your-tenant-name>.
- Use the name and tag for the container that you pushed in the previous step.
Click “Start service” if the service definition looks good to you.

Service definition
{
  "name": "python-publish-to-stream",
  "image": "registry.cp.kpn-dsh.com/<your-tenant-name>/python-publish-stream:1.0.0",
  "cpus": 0.1,
  "mem": 256,
  "env": {
    "PRODUCE_STREAM_TOPIC": "<tenant-topic-in-public-stream>",
    "MQTT_TOPIC": "house/kitchen/sensor"
  },
  "instances": 1,
  "singleInstance": false,
  "needsToken": true,
  "user": "<tenant-user-ID>"
}

The Python script uses the environment variables in the service definition (line 6–9):

PRODUCE_STREAM_TOPIC: The destination to write messages to. Make sure that you fill out the full name of your Kafka topic here.
MQTT_TOPIC: The MQTT topic to write messages to. For the demonstration, we use the MQTT topic ‘house/kitchen/sensor’, but you can change it. Make sure that your MQTT topic doesn’t have a leading forward slash (/).

Inspect the service¶

When you start the service, the DSH automatically redirects you to the details page of your service. You can also reach this page by clicking “Services” > “Overview” in the menu of the DSH Console, and then clicking the relevant line for your service in the overview page.

Grafana¶

You can inspect the output of the service:

Navigate to the details page of the service if you aren’t already there.
Under “Running tasks”, click the button with the blue “Page” icon at the right of the running task.
In a new browser tab, the DSH leads you to the correct query in Grafana for your service’s logs:
- Scroll down to inspect the log entries.
- It may take a minute before log entries start coming in.
- Click the “Live” button at the top right of your Grafana page to see the log entries in real time, or you can refresh the page manually.
If all goes well, you’ll see the following messages appear:
- The output of the set_up_config.sh script
- The message <timestamp> Message delivered to <scratch-topic-address> [<kafka-partition>] @ <offset>, as defined in the Python script.

Now stop your service:

Head back to the details page of your service.
Click the “Stop” button at the top right of the page.
Go back to the log entries in Grafana. The logs should show <timestamp> Received SIGTERM, shutting down., as defined in the Python script.

Kafdrop¶

You can use Kafdrop to inspect your Kafka topic:

First, deploy the Kafdrop app if you haven’t done so already. If you already deployed Kafdrop, then you can skip the following steps:
- In the menu bar of the DSH Console, click “Services” > “App Catalog”.
- Click the title of the “Kafdrop” app, and then the “Configure & Deploy” button.
- Fill out the configuration form correctly, and click the “Deploy” button.
Once the DSH deployed your Kafdrop app, you can open it:
- Click “Services” > “Overview”, and then click the name that you chose for your Kafdrop app.
- On the details page of your Kafdrop app, click the icon next to the link under “Services & resources” to navigate to your Kafdrop app.
- Log in if necessary.
Now, you can inspect your Kafka topic:
- In Kafdrop, click the name of your Kafka topic. You can find it in the table under “Topics”, and the name has the format stream.<topic-name>.<your-tenant-name>.
- Click the “View Messages” button on the details page of your Kafka topic.
- Click the “View Messages” button again on the “Topic Messages” page. Make sure that you select the correct partition.
- Kafdrop displays the messages in your topic, with their timestamp and message content.
- You can use the buttons above the list to navigate through the list.

Cmd Line¶

You can use the Cmd Line app to inspect the complete DSH stream:

First, deploy the Cmd Line app if you haven’t done so already. If you already deployed Cmd Line, then you can skip the following steps:
- In the menu bar of the DSH Console, click “Services” > “App Catalog”.
- Click the title of the “Cmd Line” app, and then the “Configure & Deploy” button.
- Fill out the configuration form correctly, and click the “Deploy” button.
Once the DSH deployed your Cmd Line app, you can open it:
- Click “Services” > “Overview”, and then click the name that you chose for your Cmd Line app.
- On the details page of your Cmd Line app, click the icon next to the link under “Services & resources” to navigate to your Cmd Line app.
- Log in if necessary.
Now, you can inspect your public DSH stream:
- Enter the command dshkcl consume stream.<public-DSH-stream-name>.* | jq .
- The Cmd Line app will display all the messages in the public DSH stream, in the JSON format.
- Check the documentation at the top of the Cmd Line app to discover other commands.

Congratulations: you have deployed your first Kafka producer for a public DSH stream, in Python. Next, you can deploy a Kafka consumer of a public DSH stream in Python.

Publish to public DSH stream¶

Prerequisites¶

Key concepts¶

Public DSH stream¶

Messaging API¶

Message envelopes¶

Create your files¶

Working directory¶

Bash scripts¶

Bash script for configuration¶

Bash script for entrypoint¶

Protobuf message envelopes¶

Python script¶

Dockerfile¶

Build the image¶

Log in to the DSH container image registry¶

Build and push the container image¶

Deploy the custom service¶

Inspect the service¶

Grafana¶

Kafdrop¶

Cmd Line¶

Further reading¶