Skip to content

Publish to Kafka topic of type “scratch”

This page describes the steps to publish messages to a Kafka topic of type “scratch” on the DSH. You will create a very simple Kafka producer in Python that sends messages to a Kafka topic on the DSH’s Kafka cluster 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 Kafka topic of type “scratch” in that tenant. See Adding a topic for more information.
    • 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

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 scratch-topic-publish
cd scratch-topic-publish

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
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
#!/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 "$@"

Python script

This script does the actual work of producing messages to the Kafka topic:

  • It retrieves the configuration for the SSL connection and the Kafka cluster.
  • It creates a Producer object that publishes messages to the Kafka topic every 5 seconds.
  • It also creates a log entry to register the success or failure to publish messages.
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import os
import sys
import json
import time
import signal
import uuid
from confluent_kafka import Producer

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

# Retrieve the configuration that the set_up_config.sh set.
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.
    tenant_cfg = os.getenv("JSON_TENANT_CONFIG")
    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()))


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, and retrieve the Kafka topic (of type "scratch") from the environment variables.
    producer = Producer(config)
    topic = os.environ.get("PRODUCE_TOPIC")
    if not topic:
        print("Error: PRODUCE_TOPIC not set.")
        sys.exit(1)

    # Create a random string, and publish it to the Kafka topic. Repeat every 5 seconds.
    count = 0
    while True:
        some_string = uuid.uuid4().hex
        value = f"message-{count}-{some_string}"
        try:
            producer.produce(topic, value, callback=delivery_callback)
            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 10–12 and line 54).
  • The script logs results to standard error output for the sake of demonstration (line 45–49). However, this isn’t a recommended way of working if you deploy custom services to a production environment because this is too chatty.

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
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 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 

# 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.

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-scratch:1.0.0 .
docker push registry.cp.kpn-dsh.com/<your-tenant-name>/python-publish-scratch: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-scratch’.
  • 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.
    • <topic-name>: The name of the Kafka topic of type “scratch” that you created for this tutorial
    • 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
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
{
  "name": "python-publish-to-scratch",
  "image": "registry.cp.kpn-dsh.com/<your-tenant-name>/python-publish-scratch:1.0.0",
  "cpus": 0.1,
  "mem": 256,
  "env": {
    "PRODUCE_TOPIC": "{ topic('<topic-name>') }"
  },
  "instances": 1,
  "singleInstance": false,
  "needsToken": true,
  "user": "<tenant-user-ID>"
  "topics": [
    "<topic-name>"
  ] 
}

Some aspects of this script are worth noting:

  • The Python script uses the environment variable PRODUCE_TOPIC (line 7) as the destination to write messages to. Make sure that you fill out the full name of your Kafka topic here, in the format scratch.<topic-name>.<your-tenant-name>.
  • The topics key (line 13-15) is optional, but it ensures that the Kafka topic is available on the DSH when you deploy your custom service. See Kafka topics for more information.

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 also 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’ll find it in the table under “Topics”, and the name has the format scratch.<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.
    • 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.

Congratulations: you have deployed your first Kafka producer in Python. Next, you can deploy a Kafka consumer in Python.

Further reading

Check out the following resources to find out more about Kafka producers in Python: