/deepstream-dev

Kafka and Message Broker Integration

Overview

This document is a comprehensive reference for integrating DeepStream applications with external message brokers. It covers two complementary areas:

• Part 1 -- Kafka Integration Use Cases and Patterns: Pipeline architectures for streaming analytics data to Apache Kafka, including native nvmsgbroker pipelines, Python Kafka producer probes, multi-topic integration, error handling, and performance optimization.
• Part 2 -- Message Broker and Converter Configuration Reference: Detailed property tables and configuration file formats for the nvmsgconv and nvmsgbroker GStreamer plugins, protocol adaptor libraries (Kafka, MQTT, Redis, AMQP, Azure IoT), payload schemas, and troubleshooting guidance.

Part 1: Kafka Integration Use Cases and Patterns

Use Case Requirements

• Process video streams with AI inference
• Extract object detection and tracking metadata
• Stream metadata to Kafka topics
• Support multiple Kafka topics for different data types
• Handle Kafka connection failures gracefully
• Support both sync and async message sending
• Integrate with cloud services and data pipelines

Prerequisites

Before building any Kafka-based DeepStream pipeline, install these system dependencies:

# REQUIRED: librdkafka -- DeepStream's Kafka protocol adapter (libnvds_kafka_proto.so)
# dynamically links against librdkafka.so.1, which is NOT bundled with DeepStream.
sudo apt-get install -y librdkafka-dev

# If also running a local MQTT broker for tracker:
sudo apt-get install -y libmosquitto1        # client library for nvtracker
sudo apt-get install -y mosquitto            # broker daemon (if running locally)
sudo apt-get install -y mosquitto-clients    # CLI tools for testing

Without librdkafka-dev, any pipeline using nvmsgbroker with the Kafka protocol adapter will fail at startup with: unable to open shared library / Failed to start.

Architecture Overview

Critical Rule: async=0 on ALL Sinks

CRITICAL: When using tee to split a pipeline OR using dynamic sources (nvmultiurisrcbin), ALL sink elements MUST have async: 0. This includes:

• Display sinks (nveglglessink, nv3dsink)
• Message broker sinks (nvmsgbroker)
• File sinks (filesink)
• Any other sink element

Symptom if missing: Pipeline stays stuck in PAUSED state. Cameras show "added" but no video displays and no data flows.

Why: GStreamer requires all sinks to "preroll" (receive data) before transitioning to PLAYING state. With async: 0, sinks don't block the state transition waiting for preroll.

Pipeline Architecture

IMPORTANT: nvmsgbroker is a SINK component that terminates the pipeline branch. It cannot have downstream components.

For headless pipelines (Kafka only, no display):

Source -> Decoder -> Muxer -> Inference -> Tracker -> Message Converter -> Message Broker (sink)

For pipelines with both Kafka and display, use tee to split paths:

Source -> Decoder -> Muxer -> Inference -> Tracker -> Tee
                                                      |-> [Metadata Branch] Message Converter -> Message Broker (sink)
                                                      |-> [Video Branch] Tiler -> OSD -> Converter -> Renderer (sink)

Data Flow

1. Video processing generates metadata (objects, tracks, frames)
2. Metadata is converted to message format
3. Messages are sent to Kafka broker (metadata branch terminates here)
4. Video continues to display pipeline (if using tee split)
5. Downstream Kafka consumers process analytics data

Implementation Approaches

Approach 1: Using nvmsgbroker Plugin (Native DeepStream)

The native DeepStream approach uses nvmsgbroker plugin with Kafka protocol library.

CRITICAL: nvmsgbroker is a SINK component that terminates the pipeline branch. It cannot have downstream components like OSD or renderer. If you need both Kafka output and display, use tee to split the pipeline into separate branches.

For detailed property tables and configuration file formats for nvmsgconv and nvmsgbroker, see Part 2 below.

Example 1: Headless Pipeline (Kafka Only)

from pyservicemaker import Pipeline
import platform
import sys

def kafka_native_pipeline_headless(video_path, infer_config, kafka_config):
    """
    DeepStream pipeline with native Kafka integration (headless, no display)

    Args:
        video_path: Path to video file
        infer_config: Inference configuration file
        kafka_config: Kafka configuration dict
    """
    pipeline = Pipeline("kafka-pipeline-headless")

    # Source and decoding
    pipeline.add("filesrc", "src", {"location": video_path})
    pipeline.add("h264parse", "parser")
    pipeline.add("nvv4l2decoder", "decoder")
    pipeline.add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})

    # Inference
    pipeline.add("nvinfer", "pgie", {"config-file-path": infer_config})

    # Tracker
    pipeline.add("nvtracker", "tracker", {
        "ll-lib-file": "/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so",
        "ll-config-file": "/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_tracker_NvDCF_perf.yml"
    })

    # Message converter (converts metadata to message format)
    # IMPORTANT: msg2p-newapi=True uses NvDsObjectMeta directly (no NvDsEventMsgMeta required)
    pipeline.add("nvmsgconv", "msgconv", {
        "config": kafka_config["msgconv_config"],
        "payload-type": 0,  # 0=deepstream full schema, 1=minimal
        "msg2p-newapi": True,  # CRITICAL: Use new API to avoid NvDsEventMsgMeta requirement
    })

    # Message broker (Kafka) - THIS IS A SINK, terminates the pipeline
    # IMPORTANT: conn-str uses semicolon separator (host;port), NOT colon
    pipeline.add("nvmsgbroker", "msgbroker", {
        "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
        "conn-str": kafka_config["broker_servers"],  # Must be "host;port" format
        "sync": 0,   # 0=async message sending, 1=sync
        "async": 0,  # CRITICAL for dynamic sources: prevents state transition deadlock
        "config": kafka_config["broker_config"]
    })

    # Link pipeline - msgbroker is the sink, no components after it
    pipeline.link("src", "parser", "decoder")
    pipeline.link(("decoder", "mux"), ("", "sink_%u"))
    pipeline.link("mux", "pgie", "tracker", "msgconv", "msgbroker")

    pipeline.start().wait()

Example 2: Pipeline with Both Kafka and Display (Using Tee)

from pyservicemaker import Pipeline
import platform
import sys

def kafka_native_pipeline_with_display(video_path, infer_config, kafka_config):
    """
    DeepStream pipeline with native Kafka integration AND display

    Uses tee to split pipeline into metadata branch (Kafka) and video branch (display)

    Args:
        video_path: Path to video file
        infer_config: Inference configuration file
        kafka_config: Kafka configuration dict
    """
    pipeline = Pipeline("kafka-pipeline-with-display")

    # Source and decoding
    pipeline.add("filesrc", "src", {"location": video_path})
    pipeline.add("h264parse", "parser")
    pipeline.add("nvv4l2decoder", "decoder")
    pipeline.add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})

    # Inference
    pipeline.add("nvinfer", "pgie", {"config-file-path": infer_config})

    # Tracker
    pipeline.add("nvtracker", "tracker", {
        "ll-lib-file": "/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so",
        "ll-config-file": "/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_tracker_NvDCF_perf.yml"
    })

    # Add tee to split pipeline
    pipeline.add("tee", "tee")

    # Metadata branch: tee -> queue -> msgconv -> msgbroker (sink)
    pipeline.add("queue", "queue_meta")
    # IMPORTANT: msg2p-newapi=True uses NvDsObjectMeta directly (no NvDsEventMsgMeta required)
    pipeline.add("nvmsgconv", "msgconv", {
        "config": kafka_config["msgconv_config"],
        "payload-type": 0,
        "msg2p-newapi": True,  # CRITICAL: Use new API
    })
    # IMPORTANT: conn-str uses semicolon separator (host;port), NOT colon
    # CRITICAL: async=0 required on ALL sinks when using tee or dynamic sources!
    pipeline.add("nvmsgbroker", "msgbroker", {
        "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
        "conn-str": kafka_config["broker_servers"],  # Must be "host;port" format
        "sync": 0,   # Async message sending
        "async": 0,  # CRITICAL: ALL sinks need async=0 to prevent state deadlock!
        "config": kafka_config["broker_config"]
    })

    # Video branch: tee -> queue -> tiler -> osd -> converter -> sink
    pipeline.add("queue", "queue_video")
    pipeline.add("nvmultistreamtiler", "tiler", {"rows": 1, "columns": 1})
    pipeline.add("nvosdbin", "osd")
    pipeline.add("nvvideoconvert", "converter")
    sink_type = "nv3dsink" if platform.processor() == "aarch64" else "nveglglessink"
    # CRITICAL: async=0 required on ALL sinks when using tee or dynamic sources!
    pipeline.add(sink_type, "sink", {
        "sync": 0,   # Don't sync to clock for live sources
        "qos": 0,    # Disable QoS
        "async": 0   # CRITICAL: ALL sinks need async=0 to prevent state deadlock!
    })

    # Link main pipeline
    pipeline.link("src", "parser", "decoder")
    pipeline.link(("decoder", "mux"), ("", "sink_%u"))
    pipeline.link("mux", "pgie", "tracker", "tee")

    # Link metadata branch (terminates at msgbroker sink)
    pipeline.link(("tee", "queue_meta"), ("src_%u", ""))
    pipeline.link("queue_meta", "msgconv", "msgbroker")

    # Link video branch (terminates at display sink)
    pipeline.link(("tee", "queue_video"), ("src_%u", ""))
    pipeline.link("queue_video", "tiler", "osd", "converter", "sink")

    pipeline.start().wait()

if __name__ == "__main__":
    kafka_config = {
        # IMPORTANT: Use semicolon separator, NOT colon!
        "broker_servers": "localhost;9092",  # Correct: semicolon
        # "broker_servers": "localhost:9092",  # Wrong: colon
        "broker_config": "/path/to/kafka_broker_config.txt",
        "msgconv_config": "/path/to/msgconv_config.txt"
    }
    # Use headless version for Kafka-only, or with_display version for both Kafka and display
    kafka_native_pipeline_headless(sys.argv[1], sys.argv[2], kafka_config)
    # OR
    # kafka_native_pipeline_with_display(sys.argv[1], sys.argv[2], kafka_config)

Example 3: Using Legacy API (msg2p-newapi=0) with EventMessageUserMetadata

When msg2p-newapi is 0 (the default), nvmsgconv expects NvDsEventMsgMeta to be pre-attached to each frame buffer. This metadata is NOT generated automatically by any DeepStream plugin. You must attach it via a probe upstream of nvmsgconv.

There are two sub-approaches:

Option A: Built-in add_message_meta_probe (Simplest)

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator
import platform

def kafka_legacy_builtin_probe(video_path, infer_config, kafka_config):
    """
    Kafka pipeline using msg2p-newapi=0 with built-in add_message_meta_probe.
    The built-in probe automatically generates EventMessageUserMetadata
    from NvDsObjectMeta for every detected object.
    """
    pipeline = Pipeline("kafka-legacy-builtin")

    # Source and decoding
    pipeline.add("filesrc", "src", {"location": video_path})
    pipeline.add("h264parse", "parser")
    pipeline.add("nvv4l2decoder", "decoder")
    pipeline.add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})

    # Inference + tracker
    pipeline.add("nvinfer", "pgie", {"config-file-path": infer_config})
    pipeline.add("nvtracker", "tracker", {
        "ll-lib-file": "/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so",
        "ll-config-file": "/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_tracker_NvDCF_perf.yml"
    })

    # OSD (needed as attachment point for the built-in probe)
    pipeline.add("nvosdbin", "osd")

    # Tee to split display and Kafka branches
    pipeline.add("tee", "tee")

    # Metadata branch
    pipeline.add("queue", "queue_meta")
    pipeline.add("nvmsgconv", "msgconv", {
        "config": kafka_config["msgconv_config"],
        "payload-type": 0,
        "msg2p-newapi": 0,  # Legacy API - requires EventMessageUserMetadata
    })
    pipeline.add("nvmsgbroker", "msgbroker", {
        "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
        "conn-str": kafka_config["broker_servers"],
        "sync": 0,
        "async": 0,
    })

    # Display branch
    pipeline.add("queue", "queue_video")
    sink_type = "nv3dsink" if platform.processor() == "aarch64" else "nveglglessink"
    pipeline.add(sink_type, "sink", {"sync": 0, "qos": 0, "async": 0})

    # Link
    pipeline.link("src", "parser", "decoder")
    pipeline.link(("decoder", "mux"), ("", "sink_%u"))
    pipeline.link("mux", "pgie", "tracker", "osd", "tee")
    pipeline.link(("tee", "queue_meta"), ("src_%u", ""))
    pipeline.link("queue_meta", "msgconv", "msgbroker")
    pipeline.link(("tee", "queue_video"), ("src_%u", ""))
    pipeline.link("queue_video", "sink")

    # CRITICAL: attach built-in probe AFTER osd, BEFORE tee->msgconv
    # This automatically creates EventMessageUserMetadata from NvDsObjectMeta
    pipeline.attach("osd", "add_message_meta_probe", "metadata generator")

    pipeline.start().wait()

Reference: deepstream_test4_app sample (/opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test4_app/deepstream_test4.py)

Option B: Custom EventMessageGenerator (Multi-Camera / Custom Sensor Mappings)

For multi-camera pipelines where you need control over sensor IDs and URIs:

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator, SensorInfo

class EventMessageGenerator(BatchMetadataOperator):
    """
    Generate EventMessageUserMetadata for downstream nvmsgconv.
    Required when msg2p-newapi=0 (legacy API).

    Uses pyservicemaker API:
        batch_meta.acquire_event_message_meta()  -> acquire from pool
        event_msg.generate(obj, frame, sensor_id, uri, labels)  -> populate
        frame_meta.append(event_msg)  -> attach to frame
    """

    def __init__(self, sensor_map, labels):
        super().__init__()
        self._sensor_map = sensor_map  # dict: source_id (int) -> SensorInfo
        self._labels = labels          # list of class label strings

    def handle_metadata(self, batch_meta, frame_interval=1):
        for frame_meta in batch_meta.frame_items:
            frame_num = frame_meta.frame_number
            for object_meta in frame_meta.object_items:
                if not (frame_num % frame_interval):
                    event_msg = batch_meta.acquire_event_message_meta()
                    if event_msg:
                        source_id = frame_meta.source_id
                        sensor_info = self._sensor_map.get(source_id)
                        sensor_id = sensor_info.sensor_id if sensor_info else "N/A"
                        uri = sensor_info.uri if sensor_info else "N/A"
                        event_msg.generate(
                            object_meta, frame_meta, sensor_id, uri, self._labels
                        )
                        frame_meta.append(event_msg)


def kafka_legacy_custom_generator(video_paths, infer_config, kafka_config, labels):
    """
    Multi-camera Kafka pipeline using msg2p-newapi=0 with custom EventMessageGenerator.
    """
    pipeline = Pipeline("kafka-legacy-custom")

    # Build sensor map from video paths
    sensor_map = {}
    for i, uri in enumerate(video_paths):
        sensor_map[i] = SensorInfo(
            sensor_id=f"Camera{i+1}",
            sensor_name=f"cam{i+1}",
            uri=uri
        )

    # ... (add sources, inference, tracker, tee, msgconv with msg2p-newapi=0, etc.)

    # Attach custom EventMessageGenerator probe UPSTREAM of nvmsgconv
    pipeline.attach(
        "tracker",
        Probe("event_msg_gen", EventMessageGenerator(sensor_map, labels))
    )

    pipeline.start().wait()

Key API calls:

• batch_meta.acquire_event_message_meta() -- acquires EventMessageUserMetadata from the pool
• event_msg.generate(object_meta, frame_meta, sensor_id, uri, labels) -- populates the metadata
• frame_meta.append(event_msg) -- attaches it to the frame for downstream nvmsgconv

Reference: deepstream_test5_app sample (/opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test5_app/deepstream_test5.py)

Kafka Broker Configuration File

kafka_broker_config.txt:

[broker]
enable=1
broker-ip-port=localhost:9092
topic=deepstream-analytics
# Optional: SSL/TLS configuration
# enable-tls=1
# ca-file=/path/to/ca-cert
# client-cert-file=/path/to/client-cert
# client-key-file=/path/to/client-key

Message Converter Configuration File

msgconv_config.txt:

[message-converter]
enable=1
# Message format: deepstream or custom
msg-format=deepstream
# Schema file for custom format
schema-file=/path/to/schema.json
# Payload type: 0=deepstream, 1=custom
payload-type=0

Approach 2: Using Python Kafka Producer (Custom Probe)

This approach uses Python's kafka-python library in a custom probe for more control.

Custom Kafka Producer Probe

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator
from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
import sys
import platform

class KafkaMetadataSender(BatchMetadataOperator):
    """
    Custom probe to send metadata to Kafka

    Sends object detection and tracking metadata to Kafka topics
    """
    def __init__(self, kafka_config):
        """
        Initialize Kafka producer

        Args:
            kafka_config: Dict with Kafka configuration
                - bootstrap_servers: Kafka broker addresses
                - topic: Topic name
                - security_config: Optional security config
        """
        super().__init__()

        # Kafka producer configuration
        producer_config = {
            "bootstrap_servers": kafka_config["bootstrap_servers"],
            "value_serializer": lambda v: json.dumps(v).encode('utf-8'),
            "key_serializer": lambda k: str(k).encode('utf-8') if k else None,
            "acks": "all",  # Wait for all replicas
            "retries": 3,
            "max_in_flight_requests_per_connection": 1,
            "enable_idempotence": True
        }

        # Add security configuration if provided
        if "security_config" in kafka_config:
            security = kafka_config["security_config"]
            if security.get("use_ssl"):
                producer_config.update({
                    "security_protocol": "SSL",
                    "ssl_cafile": security.get("ca_file"),
                    "ssl_certfile": security.get("cert_file"),
                    "ssl_keyfile": security.get("key_file")
                })
            elif security.get("use_sasl"):
                producer_config.update({
                    "security_protocol": "SASL_SSL",
                    "sasl_mechanism": security.get("sasl_mechanism", "PLAIN"),
                    "sasl_plain_username": security.get("username"),
                    "sasl_plain_password": security.get("password")
                })

        self.producer = KafkaProducer(**producer_config)
        self.topic = kafka_config["topic"]
        self.send_frame_metadata = kafka_config.get("send_frame_metadata", True)
        self.send_object_metadata = kafka_config.get("send_object_metadata", True)
        self.batch_size = kafka_config.get("batch_size", 1)  # Send every N frames

        self.frame_count = 0
        self.error_count = 0

    def handle_metadata(self, batch_meta):
        """Process batch metadata and send to Kafka"""
        for frame_meta in batch_meta.frame_items:
            self.frame_count += 1

            # Send metadata every N frames (if batch_size > 1)
            if self.frame_count % self.batch_size != 0:
                continue

            try:
                # Prepare message
                message = self._prepare_message(frame_meta)

                # Send to Kafka
                future = self.producer.send(
                    topic=self.topic,
                    key=str(frame_meta.frame_number),  # Use frame number as key
                    value=message
                )

                # Optional: Add callback for success/failure
                future.add_callback(self._on_send_success)
                future.add_errback(self._on_send_error)

            except Exception as e:
                print(f"Error sending message to Kafka: {e}")
                self.error_count += 1

    def _prepare_message(self, frame_meta):
        """Prepare message from frame metadata"""
        message = {
            "frame_number": frame_meta.frame_number,
            # Note: Use buffer_pts for PTS timestamp, ntp_timestamp for NTP timestamp
            "buffer_pts": frame_meta.buffer_pts,
            "ntp_timestamp": frame_meta.ntp_timestamp,
            "pad_index": frame_meta.pad_index,
            "source_id": frame_meta.source_id  # Use source_id property
        }

        # Add frame-level metadata
        if self.send_frame_metadata:
            message["frame_metadata"] = {
                "source_width": frame_meta.source_width,
                "source_height": frame_meta.source_height,
                "pipeline_width": frame_meta.pipeline_width,
                "pipeline_height": frame_meta.pipeline_height
            }

        # Add object metadata
        if self.send_object_metadata:
            objects = []
            for obj_meta in frame_meta.object_items:
                obj_data = {
                    "class_id": obj_meta.class_id,
                    "confidence": float(obj_meta.confidence),
                    # Use object_id to get the tracker-assigned tracking ID
                    "object_id": obj_meta.object_id,
                    "bbox": {
                        "left": float(obj_meta.rect_params.left),
                        "top": float(obj_meta.rect_params.top),
                        "width": float(obj_meta.rect_params.width),
                        "height": float(obj_meta.rect_params.height)
                    }
                }

                # Add secondary inference results if available
                # (stored in obj_meta.obj_user_meta_list)
                if hasattr(obj_meta, 'obj_user_meta_list'):
                    obj_data["attributes"] = self._extract_attributes(obj_meta)

                objects.append(obj_data)

            message["objects"] = objects
            message["object_count"] = len(objects)

        return message

    def _extract_attributes(self, obj_meta):
        """Extract secondary inference attributes from object metadata"""
        attributes = {}
        # Process obj_user_meta_list to extract classification results
        # This depends on how secondary inference stores results
        return attributes

    def _on_send_success(self, record_metadata):
        """Callback for successful message send"""
        pass  # Can add logging here

    def _on_send_error(self, exception):
        """Callback for failed message send"""
        print(f"Kafka publish failed: {exception}")
        self.error_count += 1

    def flush(self):
        """Flush pending messages"""
        self.producer.flush()

    def close(self):
        """Close Kafka producer"""
        self.producer.flush()
        self.producer.close()
        print(f"Kafka producer closed. Sent {self.frame_count} frames, {self.error_count} errors")

def kafka_custom_probe_pipeline(video_path, infer_config, kafka_config):
    """Pipeline with custom Kafka probe"""
    pipeline = Pipeline("kafka-custom-probe")

    # Source and decoding
    pipeline.add("filesrc", "src", {"location": video_path})
    pipeline.add("h264parse", "parser")
    pipeline.add("nvv4l2decoder", "decoder")
    pipeline.add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})

    # Inference
    pipeline.add("nvinfer", "pgie", {"config-file-path": infer_config})

    # Tracker
    pipeline.add("nvtracker", "tracker", {
        "ll-lib-file": "/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so",
        "ll-config-file": "/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_tracker_NvDCF_perf.yml"
    })

    # OSD and sink
    pipeline.add("nvosdbin", "osd")
    pipeline.add("nvvideoconvert", "converter")
    sink_type = "nv3dsink" if platform.processor() == "aarch64" else "nveglglessink"
    pipeline.add(sink_type, "sink", {"sync": 1})

    # Link pipeline
    pipeline.link("src", "parser", "decoder")
    pipeline.link(("decoder", "mux"), ("", "sink_%u"))
    pipeline.link("mux", "pgie", "tracker", "osd", "converter", "sink")

    # Attach Kafka probe
    kafka_sender = KafkaMetadataSender(kafka_config)
    pipeline.attach("tracker", Probe("kafka-sender", kafka_sender))

    try:
        pipeline.start().wait()
    finally:
        kafka_sender.close()

if __name__ == "__main__":
    kafka_config = {
        "bootstrap_servers": "localhost:9092",
        "topic": "deepstream-analytics",
        "send_frame_metadata": True,
        "send_object_metadata": True,
        "batch_size": 1  # Send every frame
    }
    kafka_custom_probe_pipeline(sys.argv[1], sys.argv[2], kafka_config)

Approach 3: Multi-Topic Kafka Integration

Send different types of metadata to different Kafka topics.

class MultiTopicKafkaSender(BatchMetadataOperator):
    """Send different metadata types to different Kafka topics"""
    def __init__(self, kafka_configs):
        """
        Args:
            kafka_configs: Dict mapping topic names to Kafka configs
                {
                    "object-detections": {...},
                    "tracking-events": {...},
                    "frame-metadata": {...}
                }
        """
        super().__init__()
        self.producers = {}
        self.topics = {}

        for topic_name, config in kafka_configs.items():
            producer = KafkaProducer(
                bootstrap_servers=config["bootstrap_servers"],
                value_serializer=lambda v: json.dumps(v).encode('utf-8')
            )
            self.producers[topic_name] = producer
            self.topics[topic_name] = config.get("topic", topic_name)

    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # Send object detections
            if "object-detections" in self.producers:
                detections = self._prepare_detections(frame_meta)
                self.producers["object-detections"].send(
                    topic=self.topics["object-detections"],
                    value=detections
                )

            # Send tracking events (new tracks, lost tracks)
            if "tracking-events" in self.producers:
                events = self._prepare_tracking_events(frame_meta)
                if events:
                    self.producers["tracking-events"].send(
                        topic=self.topics["tracking-events"],
                        value=events
                    )

            # Send frame metadata
            if "frame-metadata" in self.producers:
                frame_data = self._prepare_frame_metadata(frame_meta)
                self.producers["frame-metadata"].send(
                    topic=self.topics["frame-metadata"],
                    value=frame_data
                )

    def _prepare_detections(self, frame_meta):
        """Prepare object detection message"""
        # Build detections list by iterating (object_items is an iterator)
        detections = [
            {
                "class_id": obj.class_id,
                "confidence": float(obj.confidence),
                "bbox": {
                    "left": float(obj.rect_params.left),
                    "top": float(obj.rect_params.top),
                    "width": float(obj.rect_params.width),
                    "height": float(obj.rect_params.height)
                }
            }
            for obj in frame_meta.object_items
        ]
        return {
            "frame_number": frame_meta.frame_number,
            "buffer_pts": frame_meta.buffer_pts,  # Use buffer_pts for timestamp
            "ntp_timestamp": frame_meta.ntp_timestamp,
            "detections": detections
        }

    def _prepare_tracking_events(self, frame_meta):
        """Prepare tracking event message"""
        # Detect new tracks, lost tracks, etc.
        # This requires maintaining state across frames
        return {}  # Implement tracking event detection

    def _prepare_frame_metadata(self, frame_meta):
        """Prepare frame metadata message"""
        # Note: object_items is an ITERATOR, not a list - cannot use len() directly
        # Count objects by iterating
        obj_count = sum(1 for _ in frame_meta.object_items)
        return {
            "frame_number": frame_meta.frame_number,
            "buffer_pts": frame_meta.buffer_pts,  # Use buffer_pts for timestamp
            "ntp_timestamp": frame_meta.ntp_timestamp,
            "object_count": obj_count
        }

    def close(self):
        """Close all producers"""
        for producer in self.producers.values():
            producer.flush()
            producer.close()

Error Handling and Resilience

Retry Logic and Error Handling

class ResilientKafkaSender(BatchMetadataOperator):
    """Kafka sender with retry logic and error handling"""
    def __init__(self, kafka_config):
        super().__init__()
        self.config = kafka_config
        self.max_retries = kafka_config.get("max_retries", 3)
        self.retry_delay = kafka_config.get("retry_delay", 1.0)
        self.message_queue = []  # Queue for failed messages
        self._init_producer()

    def _init_producer(self):
        """Initialize or reinitialize producer"""
        try:
            self.producer = KafkaProducer(
                bootstrap_servers=self.config["bootstrap_servers"],
                value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                retries=self.max_retries,
                max_in_flight_requests_per_connection=1,
                enable_idempotence=True
            )
            self.connected = True
        except Exception as e:
            print(f"Failed to initialize Kafka producer: {e}")
            self.connected = False

    def handle_metadata(self, batch_meta):
        if not self.connected:
            self._init_producer()
            if not self.connected:
                # Store messages for later retry
                self.message_queue.append(batch_meta)
                return

        try:
            # Process current batch
            self._send_batch(batch_meta)

            # Retry queued messages
            while self.message_queue:
                queued_batch = self.message_queue.pop(0)
                try:
                    self._send_batch(queued_batch)
                except Exception as e:
                    # Re-queue if still failing
                    self.message_queue.append(queued_batch)
                    break

        except Exception as e:
            print(f"Error sending to Kafka: {e}")
            self.message_queue.append(batch_meta)
            # Try to reconnect
            self.connected = False

    def _send_batch(self, batch_meta):
        """Send batch metadata to Kafka"""
        for frame_meta in batch_meta.frame_items:
            message = self._prepare_message(frame_meta)
            future = self.producer.send(
                topic=self.config["topic"],
                value=message
            )
            # Wait for delivery (synchronous for reliability)
            future.get(timeout=10)

Performance Optimization

Batching Messages

class BatchedKafkaSender(BatchMetadataOperator):
    """Batch multiple frames before sending to Kafka"""
    def __init__(self, kafka_config, batch_size=10):
        super().__init__()
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_config["bootstrap_servers"],
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            batch_size=16384,  # Kafka batch size in bytes
            linger_ms=100  # Wait up to 100ms to batch
        )
        self.topic = kafka_config["topic"]
        self.batch_size = batch_size
        self.frame_buffer = []

    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            self.frame_buffer.append(frame_meta)

            if len(self.frame_buffer) >= self.batch_size:
                self._send_batch()

    def _send_batch(self):
        """Send batched frames"""
        batch_message = {
            "frames": [self._prepare_message(f) for f in self.frame_buffer]
        }
        self.producer.send(topic=self.topic, value=batch_message)
        self.frame_buffer.clear()

    def flush(self):
        """Flush remaining frames"""
        if self.frame_buffer:
            self._send_batch()
        self.producer.flush()

Testing and Validation

Test Kafka Consumer

from kafka import KafkaConsumer
import json

def test_kafka_consumer(bootstrap_servers, topic):
    """Test consumer to verify messages are being sent"""
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=bootstrap_servers,
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        auto_offset_reset='earliest',
        enable_auto_commit=True
    )

    print(f"Consuming messages from topic: {topic}")
    for message in consumer:
        print(f"Received: {message.value}")

Common Patterns

Pattern 1: Real-time Analytics Dashboard

• Send object counts and statistics to Kafka
• Dashboard consumes and displays in real-time

Pattern 2: Data Lake Ingestion

• Send all metadata to Kafka
• Kafka Connect streams to data lake (S3, HDFS)

Pattern 3: Alert System

• Send only significant events (intrusions, anomalies)
• Alert service consumes and triggers notifications

Pattern 4: Multi-Tenant Analytics

• Use different topics for different customers/streams
• Enable topic-based access control

Part 2: Message Broker and Converter Configuration Reference

Architecture

Pipeline -> nvmsgconv -> nvmsgbroker -> External Broker
              |              |
              |              +-- Protocol Adaptor Library
              |                   (libnvds_kafka_proto.so, etc.)
              |
              +-- Config File (sensor, place, analytics metadata)

IMPORTANT: nvmsgbroker is a SINK component that terminates the pipeline branch. It cannot have downstream components.

nvmsgconv Plugin

Purpose

Converts DeepStream metadata (NvDsEventMsgMeta or NvDsFrameMeta/NvDsObjectMeta) to message payload format.

GStreamer Properties

CRITICAL: msg2p-newapi Property

Problem: By default (msg2p-newapi: false), nvmsgconv requires NvDsEventMsgMeta (exposed as EventMessageUserMetadata in pyservicemaker) to be attached to the buffer. This metadata is NOT automatically generated by inference or tracker plugins. Without explicitly handling this, nvmsgconv silently produces zero messages.

Two Solutions (pick one):

Solution A: Set msg2p-newapi=True (Simple, Recommended for Most Cases)

Uses the new API that reads directly from NvDsFrameMeta and NvDsObjectMeta without requiring NvDsEventMsgMeta:

# CORRECT - Uses object metadata directly, no NvDsEventMsgMeta needed
pipeline.add("nvmsgconv", "msgconv", {
    "config": msgconv_config,
    "payload-type": 0,
    "msg2p-newapi": True,      # Use new API - reads from NvDsObjectMeta directly
})

Solution B: Keep msg2p-newapi=0 and Attach EventMessageUserMetadata Probe

Required when using custom msg2p-lib payload libraries that expect legacy NvDsEventMsgMeta, or when you need fine-grained control over per-object message generation.

Option B1: Built-in probe (simplest):

pipeline.add("nvmsgconv", "msgconv", {
    "config": msgconv_config,
    "payload-type": 0,
    # msg2p-newapi defaults to 0 (legacy API)
})

# Built-in probe auto-generates EventMessageUserMetadata from NvDsObjectMeta
pipeline.attach("osd", "add_message_meta_probe", "metadata generator")

Option B2: Custom EventMessageGenerator (for multi-camera / custom sensor mappings):

from pyservicemaker import Probe, BatchMetadataOperator, SensorInfo

class EventMessageGenerator(BatchMetadataOperator):
    def __init__(self, sensor_map, labels):
        super().__init__()
        self._sensor_map = sensor_map  # dict: source_id -> SensorInfo
        self._labels = labels          # list of class label strings

    def handle_metadata(self, batch_meta, frame_interval=1):
        for frame_meta in batch_meta.frame_items:
            for object_meta in frame_meta.object_items:
                event_msg = batch_meta.acquire_event_message_meta()
                if event_msg:
                    source_id = frame_meta.source_id
                    sensor_info = self._sensor_map.get(source_id)
                    sensor_id = sensor_info.sensor_id if sensor_info else "N/A"
                    uri = sensor_info.uri if sensor_info else "N/A"
                    event_msg.generate(
                        object_meta, frame_meta, sensor_id, uri, self._labels
                    )
                    frame_meta.append(event_msg)

# Attach UPSTREAM of nvmsgconv (e.g., on tracker or osd element)
sensor_map = {0: SensorInfo("Camera1", "cam1", "file:///video.mp4")}
labels = ["car", "bicycle", "person", "roadsign"]
pipeline.attach("tracker", Probe("event_msg_gen", EventMessageGenerator(sensor_map, labels)))

For complete pipeline examples using the legacy API, see Part 1 above (Example 3).

Common Mistake

# WRONG - Without msg2p-newapi=True AND without EventMessageUserMetadata probe,
# nvmsgconv has no input and produces ZERO messages silently!
pipeline.add("nvmsgconv", "msgconv", {
    "config": msgconv_config,
    "payload-type": 0
})

Reference samples:

• Built-in probe: /opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test4_app/deepstream_test4.py
• Custom generator: /opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test5_app/deepstream_test5.py

Payload Types

Segmentation Payload Contract

When a consumer needs segmentation data:

1. Declare whether the payload needs per-object masks, a full-frame mask, or both; preserve normal object metadata.
2. Instance masks on object metadata are bbox-local. For a full-frame payload, give the serializer an explicit frame mask or preserved inference tensors in a declared frame grid; do not reconstruct it from tracker-updated objects unless that is the intended output.
3. If the selected schema has no mask field, use a compatible serializer extension. Payload type is a consumer contract, not a shortcut for carrying masks.
4. Decode a real payload and verify its objects, mask representation, dimensions, and coordinate space.

Note that OSD can render instance-mask metadata, but nvmsgconv does not serialize it automatically.

Pipeline Usage

# Using pyservicemaker Pipeline API
pipeline.add("nvmsgconv", "msgconv", {
    "config": "/path/to/msgconv_config.txt",
    "payload-type": 0  # Full DeepStream schema
})

nvmsgconv Configuration File

The configuration file defines metadata about sensors, places, and analytics that gets embedded in the message payload.

Supported Formats

• INI-style format (.txt) - Recommended
• YAML format (.yml)

Configuration Sections

[sensor0], [sensor1], ... - Sensor/Camera Information

[place0], [place1], ... - Location/Place Information

[analytics0], [analytics1], ... - Analytics Information

Example Configuration (INI-style)

# msgconv_config.txt

[sensor0]
enable=1
type=Camera
id=CAMERA_001
location=45.293701;-75.830391;48.155
description=Entrance Camera
coordinate=5.2;10.1;11.2

[sensor1]
enable=1
type=Camera
id=CAMERA_002
location=45.293702;-75.830392;48.156
description=Exit Camera
coordinate=6.2;11.1;12.2

[place0]
enable=1
id=1
type=garage
name=ParkingLot_A
location=30.32;-40.55;100.0
coordinate=1.0;2.0;3.0
place-sub-field1=Zone_A
place-sub-field2=Lane_1
place-sub-field3=Level_P1

[analytics0]
enable=1
id=ANALYTICS_001
description=Vehicle Detection and Tracking
source=ResNet18_TrafficCamNet
version=1.0

Example Configuration (YAML)

# msgconv_config.yml

sensor0:
  enable: 1
  type: Camera
  id: CAMERA_001
  location: 45.293701;-75.830391;48.155
  description: Entrance Camera
  coordinate: 5.2;10.1;11.2

place0:
  enable: 1
  id: 1
  type: garage
  name: ParkingLot_A
  location: 30.32;-40.55;100.0
  coordinate: 1.0;2.0;3.0
  place-sub-field1: Zone_A
  place-sub-field2: Lane_1
  place-sub-field3: Level_P1

analytics0:
  enable: 1
  id: ANALYTICS_001
  description: Vehicle Detection and Tracking
  source: ResNet18_TrafficCamNet
  version: 1.0

Multi-Source Configuration

For multi-source pipelines, create sensor/place entries for each source:

# Sensor entries map to source_id in the pipeline
[sensor0]
enable=1
type=Camera
id=STREAM_0
description=Camera 0

[sensor1]
enable=1
type=Camera
id=STREAM_1
description=Camera 1

# Place entries map to source_id
[place0]
enable=1
id=0
type=intersection
name=Location_0

[place1]
enable=1
id=1
type=intersection
name=Location_1

nvmsgbroker Plugin

Purpose

Sends payload metadata to external message brokers using protocol adaptor libraries.

GStreamer Properties

CRITICAL: async=0 for Dynamic Sources and Tee Splits

When using nvmsgbroker in a pipeline with:

• Dynamic sources (nvmultiurisrcbin)
• Tee splits (multiple branches with different sinks)

You MUST set async: 0 on nvmsgbroker AND all other sinks. Otherwise, the pipeline will be stuck in PAUSED state.

# CORRECT - async=0 for tee/dynamic source pipelines
pipeline.add("nvmsgbroker", "msgbroker", {
    "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
    "conn-str": "localhost;9092",
    "sync": 0,   # Async message sending
    "async": 0,  # CRITICAL: Required for tee/dynamic sources!
})

# WRONG - missing async=0 causes pipeline stuck in PAUSED
pipeline.add("nvmsgbroker", "msgbroker", {
    "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
    "conn-str": "localhost;9092",
    "sync": 0,
    # async defaults to 1, causing state transition deadlock!
})

Protocol Adaptor Libraries

Located at /opt/nvidia/deepstream/deepstream/lib/:

CRITICAL: Connection String Format

DeepStream message broker uses semicolon (;) as separator, NOT colon (:).

# CORRECT - semicolon separator
"conn-str": "localhost;9092"

# WRONG - colon separator (will fail to connect)
"conn-str": "localhost:9092"

Pipeline Usage

# Using pyservicemaker Pipeline API
# For simple pipelines (single source, no tee):
pipeline.add("nvmsgbroker", "msgbroker", {
    "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
    "conn-str": "localhost;9092",  # IMPORTANT: Use semicolon, not colon!
    "topic": "deepstream-analytics",
    "sync": 0,
    "config": "/path/to/kafka_config.txt"
})

# For pipelines with dynamic sources OR tee splits:
pipeline.add("nvmsgbroker", "msgbroker", {
    "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
    "conn-str": "localhost;9092",  # IMPORTANT: Use semicolon, not colon!
    "topic": "deepstream-analytics",
    "sync": 0,
    "async": 0,  # CRITICAL: Required for tee/dynamic sources!
    "config": "/path/to/kafka_config.txt"
})

Protocol Adaptor Configurations

Kafka Protocol Adaptor

Dependencies Installation

# Add Confluent repository
sudo mkdir -p /usr/share/confluent-repo
wget -qO /tmp/confluent-archive.key https://packages.confluent.io/deb/7.8/archive.key
gpg --dearmor --output /tmp/confluent.gpg /tmp/confluent-archive.key
sudo cp /tmp/confluent.gpg /usr/share/confluent-repo/confluent.gpg
sudo chmod 0644 /usr/share/confluent-repo/confluent.gpg
rm -f /tmp/confluent-archive.key /tmp/confluent.gpg

CP_DIST=$(lsb_release -cs)
cat <<EOF >/tmp/confluent-platform.sources
Types: deb
URIs: https://packages.confluent.io/deb/8.0
Suites: stable
Components: main
Architectures: $(dpkg --print-architecture)
Signed-By: /usr/share/confluent-repo/confluent.gpg

Types: deb
URIs: https://packages.confluent.io/clients/deb/
Suites: ${CP_DIST}
Components: main
Architectures: $(dpkg --print-architecture)
Signed-By: /usr/share/confluent-repo/confluent.gpg
EOF
sudo cp /tmp/confluent-platform.sources /etc/apt/sources.list.d/confluent-platform.sources
sudo chmod 0644 /etc/apt/sources.list.d/confluent-platform.sources
rm -f /tmp/confluent-platform.sources

# Install dependencies
sudo apt-get update
sudo apt-get install librdkafka-dev libglib2.0-dev libjansson-dev libssl-dev

Configuration File (cfg_kafka.txt)

[message-broker]
# Consumer group ID for Kafka consumer
#consumer-group-id = mygroup

# Generic librdkafka configuration (applies to both producer and consumer)
# Semicolon-separated key=value pairs
#proto-cfg = "message.max.bytes=200000;log_level=6"

# Producer-specific librdkafka configuration
#producer-proto-cfg = "queue.buffering.max.messages=200000;message.send.max.retries=3"

# Consumer-specific librdkafka configuration
#consumer-proto-cfg = "max.poll.interval.ms=20000"

# Partition key field name in JSON message
# Use "sensor.id" for full schema, "sensorId" for minimal schema
#partition-key = sensor.id

# Enable connection sharing within same process
#share-connection = 1

Connection String

Format: hostname;port

Example: localhost;9092 or kafka-broker.example.com;9092

TLS/SSL Configuration

For secure connections, refer to /opt/nvidia/deepstream/deepstream/sources/libs/kafka_protocol_adaptor/Security_Setup.md

MQTT Protocol Adaptor

Dependencies Installation

# Install dependencies
sudo apt-get install libglib2.0-dev libcjson-dev libssl-dev

# Add Mosquitto PPA and install
sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa
sudo apt-get update
sudo apt-get install libmosquitto-dev mosquitto

Configuration File (cfg_mqtt.txt)

[message-broker]
# Username for broker authentication (deprecated - use env var)
#username = user

# Password for broker authentication (deprecated - use env var)
#password = password

# Unique client ID (empty = random)
client-id = deepstream-client

# TLS Configuration
#enable-tls = 1
#tls-cafile = /path/to/ca-cert.pem
#tls-capath = /path/to/ca-certs-dir/
#tls-certfile = /path/to/client-cert.pem
#tls-keyfile = /path/to/client-key.pem

# Connection sharing
#share-connection = 1

# Mosquitto loop timeout in ms
#loop-timeout = 2000

# Keep-alive interval in seconds
#keep-alive = 60

# Enable threaded mode (required for nvmsgbroker plugin)
#set-threaded = 1

User Authentication via Environment Variables

export USER_MQTT=username
export PASSWORD_MQTT=password

Connection String

Format: hostname;port

Example: localhost;1883

Running Mosquitto Broker

# Add mosquitto user
sudo adduser --system mosquitto

# Run broker
mosquitto

# Or with config file
mosquitto -c /etc/mosquitto/mosquitto.conf

Verify Messages

# Subscribe to topic
mosquitto_sub -t deepstream-analytics -v

# Publish test message
mosquitto_pub -t deepstream-analytics -m 'test message'

Redis Protocol Adaptor

Dependencies Installation

# Install dependencies
sudo apt-get install libglib2.0-dev libssl-dev libhiredis-dev

Configuration File (cfg_redis.txt)

[message-broker]
# Redis server hostname
#hostname=localhost

# Redis server port
#port=6379

# Password for Redis AUTH (deprecated - use env var)
#password=password

# Redis stream key for payload
#payloadkey=metadata

# Consumer group name
#consumergroup=mygroup

# Consumer name
#consumername=myname

# Maximum stream size (for capped streams)
#streamsize=10000

# Connection sharing
#share-connection = 1

User Authentication via Environment Variables

export PASSWORD_REDIS=password

Connection String

Format: hostname;port

Example: localhost;6379

Running Redis Server

# Download and build Redis
wget http://download.redis.io/releases/redis-6.0.8.tar.gz
tar xzf redis-6.0.8.tar.gz
cd redis-6.0.8
make

# Run server
src/redis-server

# Or with protected mode disabled (for external connections)
src/redis-server --protected-mode no

AMQP Protocol Adaptor (RabbitMQ)

Dependencies Installation

# Install dependencies
sudo apt-get install libglib2.0-dev librabbitmq-dev

# Install RabbitMQ server (optional, for local testing)
sudo apt-get install rabbitmq-server
sudo service rabbitmq-server start

Configuration File (cfg_amqp.txt)

[message-broker]
# RabbitMQ server hostname
hostname = localhost

# RabbitMQ server port
port = 5672

# Username (deprecated - use env var)
username = guest

# Password (deprecated - use env var)
password = guest

# AMQP exchange name
exchange = amq.topic

# Topic/routing key
topic = deepstream-analytics

# Maximum frame size
amqp-framesize = 131072

# Heartbeat interval in seconds (0 = disabled)
#amqp-heartbeat = 0

# Connection sharing
#share-connection = 1

User Authentication via Environment Variables

export USER_AMQP=username
export PASSWORD_AMQP=password

Connection String

Format: hostname;port;username;password

Example: localhost;5672;guest;guest

Setup RabbitMQ Queue

# Enable management plugin
sudo rabbitmq-plugins enable rabbitmq_management

# Create queue
sudo rabbitmqadmin -u guest -p guest -V / declare queue name=myqueue durable=false auto_delete=true

# Bind queue to exchange
rabbitmqadmin -u guest -p guest -V / declare binding source=amq.topic destination=myqueue routing_key=deepstream-analytics

# List queues
sudo rabbitmqctl list_queues

Consume Messages

# Install amqp-tools
sudo apt-get install amqp-tools

# Consume from queue
amqp-consume -q "myqueue" -r "deepstream-analytics" -e "amq.topic" cat

Azure IoT Protocol Adaptor

Dependencies Installation

# Install dependencies
sudo apt-get update
sudo apt-get install -y libcurl4-openssl-dev libssl-dev uuid-dev libglib2.0-dev

# Build Azure IoT SDK
git clone https://github.com/Azure/azure-iot-sdk-c.git
cd azure-iot-sdk-c
git checkout tags/1.11.0
git submodule update --init

# Modify CMakeLists.txt:
# - Line 61: set build_as_dynamic to ON
# - Line 65: set use_edge_modules to ON

mkdir cmake && cd cmake
cmake ..
cmake --build .
sudo make install

Configuration File (cfg_azure.txt)

[message-broker]
# Azure IoT Hub connection string
#connection_str = HostName=<my-hub>.azure-devices.net;DeviceId=<device_id>;SharedAccessKey=<my-policy-key>

# Custom message properties (key=value pairs)
#custom_msg_properties = key1=value1;key2=value2;

# Connection sharing
#share-connection = 1

# Cleanup timeout in seconds during disconnect
#cleanup-timeout = 20

Connection String

Full Azure IoT Hub connection string:

HostName=<my-hub>.azure-devices.net;DeviceId=<device_id>;SharedAccessKey=<my-policy-key>

nvmsgbroker Library Configuration

The nvmsgbroker library (wrapper around protocol adaptors) has its own configuration:

Configuration File (cfg_nvmsgbroker.txt)

[nvmsgbroker]
# Enable auto-reconnection (0=disable, 1=enable)
auto-reconnect=1

# Reconnection retry interval in seconds
retry-interval=1

# Maximum retry limit in seconds
max-retry-limit=3600

# Work interval in microseconds
work-interval=10000

Message Payload Formats

Full Schema (payload-type=0)

Generates separate JSON payload per object:

{
  "messageid": "unique-uuid",
  "mdsversion": "1.0",
  "@timestamp": "2024-01-15T10:30:00.000Z",
  "place": {
    "id": "1",
    "name": "ParkingLot_A",
    "type": "garage",
    "location": {
      "lat": 30.32,
      "lon": -40.55,
      "alt": 100.0
    }
  },
  "sensor": {
    "id": "CAMERA_001",
    "type": "Camera",
    "description": "Entrance Camera"
  },
  "analyticsModule": {
    "id": "ANALYTICS_001",
    "description": "Vehicle Detection",
    "source": "ResNet18_TrafficCamNet",
    "version": "1.0"
  },
  "object": {
    "id": "1",
    "speed": 0,
    "direction": 0,
    "orientation": 0,
    "vehicle": {
      "type": "car",
      "make": "",
      "model": "",
      "color": "",
      "license": ""
    },
    "bbox": {
      "topleftx": 100,
      "toplefty": 200,
      "bottomrightx": 300,
      "bottomrighty": 400
    },
    "location": {
      "lat": 0,
      "lon": 0,
      "alt": 0
    },
    "coordinate": {
      "x": 0,
      "y": 0,
      "z": 0
    }
  },
  "event": {
    "id": "event-uuid",
    "type": "entry"
  },
  "videoPath": ""
}

Minimal Schema (payload-type=1)

Multiple objects in single JSON payload:

{
  "messageid": "unique-uuid",
  "mdsversion": "1.0",
  "@timestamp": "2024-01-15T10:30:00.000Z",
  "sensorId": "CAMERA_001",
  "objects": [
    {
      "id": "1",
      "type": "car",
      "confidence": 0.95,
      "bbox": {
        "topleftx": 100,
        "toplefty": 200,
        "bottomrightx": 300,
        "bottomrighty": 400
      }
    },
    {
      "id": "2",
      "type": "person",
      "confidence": 0.88,
      "bbox": {
        "topleftx": 400,
        "toplefty": 150,
        "bottomrightx": 450,
        "bottomrighty": 350
      }
    }
  ]
}

Troubleshooting

Common Issues

1. "Connection refused" error

   • Verify broker is running and accessible
   • Check firewall rules
   • Verify connection string format

2. "Library not found" error

   • Verify proto-lib path exists
   • Check library dependencies: ldd /opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so

3. Messages not appearing in broker

   • Verify topic exists (or auto-create is enabled)
   • Check broker logs for errors
   • Enable DeepStream logging (see below)

4. TLS/SSL connection failures

   • Verify certificate paths
   • Check certificate validity
   • Ensure proper permissions on key files

Enable DeepStream Logging

# Setup logger
chmod u+x <DS_ROOT>/src/utils/nvds_logger/setup_nvds_logger.sh
sudo <DS_ROOT>/src/utils/nvds_logger/setup_nvds_logger.sh

# View logs
tail -f /tmp/nvds/ds.log

Best Practices

1. Use async mode (sync=0) for better performance
2. Configure appropriate batch sizes in nvmsgconv's frame-interval
3. Use minimal schema (payload-type=1) for lower bandwidth
4. Enable auto-reconnect in nvmsgbroker config for resilience
5. Use environment variables for credentials instead of config files
6. Monitor broker lag to ensure consumers keep up
7. Use TLS/SSL for production deployments
8. Implement retry logic: Handle transient Kafka failures (see Part 1 above for Python examples)
9. Batch messages: Reduce network overhead (see Part 1 above for batching patterns)
10. Use appropriate partitioning: Use frame_number or source_id as key
11. Handle backpressure: Pause pipeline if Kafka is slow
12. Monitor producer metrics: Track send rates and errors
13. Clean shutdown: Flush and close producers properly

Related Documentation

• GStreamer Plugins Overview: gstreamer_plugins.md
• Service Maker Python API: service_maker_api.md