DEV Community: Aykut Bulgu

How to Auto-Create Jira Issues from Event Streams with Redpanda Connect

Aykut Bulgu — Mon, 17 Aug 2026 10:17:04 +0000

Your application fires hundreds of events per minute. Somewhere in that stream is a critical error that needs a Jira ticket. Someone watches a dashboard, decides the spike matters, and creates the ticket by hand. By the time the ticket exists, the incident is already twenty minutes old.

Redpanda Connect can close this gap. I wanted to see if I could get from a critical event on the topic to a ticket in Jira without writing a custom service. By combining its http processor with Bloblang mappings, you can wire your event stream directly to Jira's REST API so qualifying events create issues automatically. The filter logic, field mapping, and API call all live in a single declarative YAML file you can version-control and deploy alongside the rest of your infrastructure.

Read on to learn how I put that together, and specifically how you can:

Consume application error events from a Redpanda topic using Redpanda Connect
Filter for critical-severity events using Bloblang mappings
Shape the filtered events into Jira REST API payloads
Create Jira issues automatically using the http processor with Basic Auth
Verify the end-to-end pipeline by producing test events and checking Jira

Auto-Creating Jira Issues from Redpanda Event Streams

Suppose you work for a company that runs multiple backend services behind a shared API gateway. Each service publishes structured JSON error events to a dedicated application-errors topic on Redpanda. The events carry a severity field (info, warning, or critical), a service name, an error message, and a timestamp. The operations team wants every critical error to appear as a Jira ticket within seconds, without anyone watching a dashboard or running a script.

The pipeline does four things:

Consumes all events from the application-errors topic.
Drops everything that is not critical severity.
Transforms the remaining events into the JSON structure Jira's REST API expects.
Sends a POST request to Jira to create the issue.

Redpanda Connect keeps the routing, filtering, and transformation in one pipeline config. The http processor handles authentication and the API call, while the mapping processors handle the filtering and structural translation between your event schema and Jira's expected fields.

Redpanda Connect also includes a dedicated jira processor for querying Jira resources (searching issues by JQL, listing projects, retrieving transitions). I'm using the http processor instead because creating issues requires a write operation against the Jira REST API, which the jira processor does not support.

Prerequisites

Before you start, make sure you have the following:

A running Redpanda cluster. You can spin one up with Docker by following the Docker quickstart, or use any OS via the general quickstart. If you prefer a hosted cluster, Redpanda Serverless works as well.
Redpanda Connect installed. Run rpk connect --version to confirm. If rpk is not installed yet, see the rpk installation docs.
A Jira Cloud account with API token access. Generate a token at id.atlassian.com/manage-profile/security/api-tokens. You'll need your Jira base URL (e.g. https://your-org.atlassian.net), account email, and that token.
Python 3.8 or later with pip for running the test event producer.

Configuring the Input

The pipeline reads from Redpanda using the kafka_franz input component. Because Redpanda is Kafka API-compatible, this component works without any modification.

input:
  kafka_franz:
    seed_brokers:
      - "${REDPANDA_BROKERS:localhost:9092}"
    topics:
      - application-errors
    consumer_group: jira-error-processor

The seed_brokers field accepts a list of broker addresses. The ${REDPANDA_BROKERS:localhost:9092} syntax reads from an environment variable and falls back to localhost:9092 if the variable is not set. For Redpanda Serverless or a remote cluster, set REDPANDA_BROKERS to your bootstrap URL. The consumer_group gives this consumer its own offset tracking, so the pipeline picks up where it left off after a restart.

Filtering for Critical Events

The first mapping processor filters out events that are not critical:

pipeline:
  processors:
    - mapping: |
        root = if this.severity == "critical" {
          this
        } else {
          deleted()
        }

Calling deleted() in a Bloblang mapping drops the message from the pipeline entirely. Events with warning or info severity never reach the next processor. This keeps your Jira project clean and avoids unnecessary API calls.

Shaping the Jira API Payload

The second mapping processor transforms the error event into the JSON structure Jira's REST API expects for creating an issue:

    - mapping: |
        root.fields.project.key = "OPS"
        root.fields.summary = "CRITICAL: " + this.service + " - " + this.message
        root.fields.description = "Service: " + this.service + "\nMessage: " + this.message + "\nTimestamp: " + this.timestamp + "\nSeverity: " + this.severity
        root.fields.issuetype.name = "Bug"

Replace "OPS" with your actual Jira project key. The mapping builds a nested JSON object with fields.project.key, fields.summary, fields.description, and fields.issuetype.name, which matches Jira's v2 REST API schema for issue creation.

Calling the Jira REST API

The http processor sends the shaped payload to Jira:

    - http:
        url: "${JIRA_BASE_URL}/rest/api/2/issue"
        verb: POST
        headers:
          Content-Type: application/json
        basic_auth:
          enabled: true
          username: "${JIRA_USERNAME}"
          password: "${JIRA_API_TOKEN}"

The processor takes the current message content (the Jira payload shaped in the previous step) and sends it as the POST request body. Jira Cloud authenticates API requests using Basic Auth where the username is your Atlassian account email and the password is the API token you generated in the prerequisites. The processor replaces the message content with Jira's API response, which contains the new issue key (e.g. OPS-42).

Store all credentials in environment variables. Never hardcode tokens in YAML files.

The Complete Pipeline

Here is the full connect.yaml:

input:
  kafka_franz:
    seed_brokers:
      - "${REDPANDA_BROKERS:localhost:9092}"
    topics:
      - application-errors
    consumer_group: jira-error-processor

pipeline:
  processors:
    - mapping: |
        root = if this.severity == "critical" {
          this
        } else {
          deleted()
        }

    - mapping: |
        root.fields.project.key = "OPS"
        root.fields.summary = "CRITICAL: " + this.service + " - " + this.message
        root.fields.description = "Service: " + this.service + "\nMessage: " + this.message + "\nTimestamp: " + this.timestamp + "\nSeverity: " + this.severity
        root.fields.issuetype.name = "Bug"

    - http:
        url: "${JIRA_BASE_URL}/rest/api/2/issue"
        verb: POST
        headers:
          Content-Type: application/json
        basic_auth:
          enabled: true
          username: "${JIRA_USERNAME}"
          password: "${JIRA_API_TOKEN}"

output:
  stdout:
    codec: lines

The stdout output prints the Jira API response for each created issue. In production you would route this to a Redpanda topic or a logging aggregator instead.

Producing Test Events

The companion repository includes a Python producer at producer/produce_errors.py that publishes mock error events. Clone the repository and install the dependencies:

git clone https://github.com/draftdev/test--how-to-use-redpanda-connects-jira-processor
cd test--how-to-use-redpanda-connects-jira-processor
pip install -r requirements.txt

The producer sends three events to the application-errors topic: two with critical severity and one with warning severity.

import json
import os
from datetime import datetime, timezone
from confluent_kafka import Producer

TOPIC = "application-errors"
BROKERS = os.environ.get("REDPANDA_BROKERS", "localhost:9092")

events = [
    {
        "service": "payment-api",
        "severity": "critical",
        "message": "Connection pool exhausted",
        "timestamp": datetime.now(timezone.utc).isoformat(),
    },
    {
        "service": "auth-service",
        "severity": "warning",
        "message": "Elevated token refresh rate detected",
        "timestamp": datetime.now(timezone.utc).isoformat(),
    },
    {
        "service": "order-service",
        "severity": "critical",
        "message": "Database write timeout after 30s",
        "timestamp": datetime.now(timezone.utc).isoformat(),
    },
]

producer = Producer({"bootstrap.servers": BROKERS})
for event in events:
    producer.produce(TOPIC, json.dumps(event).encode("utf-8"))
    print(f"Sent: {event['service']} [{event['severity']}]")
producer.flush()
print("Done.")

The pipeline should create two Jira issues and discard the warning event, giving you a clear signal that the filter is working correctly.

Running the Pipeline and Verifying Output

Create the topic if it does not already exist:

rpk topic create application-errors

Export your environment variables:

export REDPANDA_BROKERS="localhost:9092"
export JIRA_BASE_URL="https://your-org.atlassian.net"
export JIRA_USERNAME="your-email@example.com"
export JIRA_API_TOKEN="your-api-token-here"

Start the pipeline:

rpk connect run connect.yaml

Connect prints its startup logs, confirms the consumer group is active, and starts waiting for messages. Keep this terminal open.

In a second terminal, run the producer:

python producer/produce_errors.py

You should see three lines in the producer terminal:

Sent: payment-api [critical]
Sent: auth-service [warning]
Sent: order-service [critical]
Done.

Back in the pipeline terminal, the Jira API responses appear for the two critical events. Each response contains the new issue key and ID. The warning event produces no output because deleted() removes it before it reaches the http processor.

To confirm the raw data that came off the topic independently of the pipeline, use rpk topic consume:

rpk topic consume application-errors --num 3

All three events print from the topic, confirming your producer and Redpanda are healthy.

Verifying the Jira Issues

Navigate to your Jira project and check the issue list. You should see two new Bug issues: one for the payment-api connection pool error and one for the order-service database write timeout.

[SCREENSHOT: Jira project showing two automatically created Bug issues with summaries matching "CRITICAL: payment-api" and "CRITICAL: order-service"]

If the issues don't appear, I'd check the pipeline terminal first for error responses from the Jira API. Some common causes are a wrong base_url format (it should not include a trailing slash), an expired API token, or a project key that does not exist in your Jira instance.

What to Build Next

I've shown you how to build a pipeline that consumes a Redpanda topic, filters events by severity, shapes each event into a Jira API payload, and creates the ticket automatically. No webhook receiver, no custom integration script, no scheduled job watching a dashboard. The whole thing lives in one YAML file, so the alerting logic sits right alongside the rest of your pipeline config instead of being a separate service to maintain.

The pattern extends further. You can add a switch processor to route critical errors from different services to different Jira projects based on the service field. You can also adjust the mapping to set priority, add labels, or assign the issue to a specific team member. For deduplication, add a cache or database lookup before the http processor to check whether an open issue already exists for the same service and error type.

Building a Custom Processor Plugin for Redpanda Connect

Aykut Bulgu — Thu, 06 Aug 2026 21:21:38 +0000

Redpanda Connect ships with a wide range of connectors and processors for building data pipelines. But, in real-world pipelines, sometimes you need transformation logic that’s not already built-in. I wanted to see how far you could take that without leaving Python, so this walks through building a dynamic plugin for Redpanda Connect.

Dynamic plugins let you extend Redpanda Connect with custom inputs, processors, or outputs written in any language that supports gRPC, running as separate subprocesses alongside the main engine. This gives you language flexibility while still using Redpanda Connect's normal pipeline config and orchestration. Your plugin is just another component in the YAML pipeline config. It loads at startup, processes messages alongside built-in components, and shuts down cleanly when the pipeline stops.

In this tutorial, I'll show you how to:

Build a custom Python processor plugin that transforms messages in a Redpanda Connect pipeline
Declare a plugin manifest and wire it into a YAML pipeline config
Run and verify the plugin end to end using rpk connect run
Extend the plugin to accept runtime configuration through the manifest

Why Use a Dynamic Plugin

Suppose most of your pipeline uses built-in Redpanda Connect processors for filtering and routing, but one step needs a proprietary text normalization routine written in Python. That routine changes often, and you don't want to recompile a Go binary every time the logic updates.

Dynamic plugins solve exactly this. You write the normalization logic as a Python function, package it alongside a plugin manifest, and load it into the pipeline at runtime with the --rpc-plugins flag. The plugin runs as a separate subprocess communicating with the main Redpanda Connect engine over gRPC. No Go code, no recompilation, no changes to the core binary.

Here's a high-level architecture diagram of a pipeline with a dynamic plugin:

Splitting the plugin into its own process has a few practical benefits:

If the plugin crashes, it doesn't take the main engine down with it. Redpanda Connect notices and restarts the subprocess.
You're not locked into Go. Any language with gRPC libraries works, and the official Python SDK handles the protocol so you're only writing the actual transformation logic.
The plugin is its own subprocess, so you can develop, test, and deploy it separately from the rest of the pipeline.

Prerequisites

You'll need these installed to follow along:

Redpanda Connect (rpk CLI, v4.56.0 or later, which includes rpk connect)
Python 3.12 or later
uv (Python package manager)

Basic familiarity with YAML pipeline configuration is helpful but not required. I'll explain each configuration field as it comes up throughout the tutorial.

Setting Up the Plugin Project

Create the plugin directory and initialize a Python project inside it:

mkdir -p plugins/yell-processor
cd plugins/yell-processor
uv init --no-readme

The uv init command creates a pyproject.toml and sets up a virtual environment. The --no-readme flag skips generating a README file since the plugin directory only needs the processor script and the manifest.

Install the Redpanda Connect Python SDK:

uv add redpanda_connect

This adds the redpanda_connect package to the project and installs it in the virtual environment. The SDK provides the decorator, message types, and gRPC server that your plugin needs to communicate with Redpanda Connect. Return to the project root before continuing:

cd ../..

The project directory should have this structure at this point:

├── plugins/
│   └── yell-processor/
│       ├── pyproject.toml
│       └── .venv/

Writing the Processor Logic

Create plugins/yell-processor/yell_processor.py with the following content:

import asyncio
import logging

import redpanda_connect


@redpanda_connect.processor
def yell(msg: redpanda_connect.Message) -> redpanda_connect.Message:
    text = msg.payload
    if isinstance(text, bytes):
        text = text.decode("utf-8")
    msg.payload = str(text).upper()
    return msg


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(redpanda_connect.processor_main(yell))

The @redpanda_connect.processor decorator marks the yell function as a processor component. The function receives a Message object and returns a modified Message. The SDK's Message is a dataclass with two main fields: payload holds the message content (bytes or a structured value), and metadata is a dictionary of key-value pairs for additional context outside the payload.

The processor_main() call at the bottom starts a gRPC server that listens on a Unix socket. Redpanda Connect launches this script as a subprocess and communicates with it through that socket. You do not need to manage the connection or protocol yourself.

Declaring the Plugin Manifest

Create plugins/yell-processor/plugin.yaml alongside the processor script:

name: yell
summary: Converts every message payload to uppercase.
command: ["uv", "run", "yell_processor.py"]
type: processor
fields: []

Each field in the manifest serves a specific purpose. name is the identifier you reference in connect.yaml when wiring the plugin into a pipeline. summary is a human-readable description of what the plugin does. command is the shell command array that Redpanda Connect executes to start the plugin subprocess, and using uv run ensures the script runs inside the project's virtual environment with all dependencies available. type declares the component type, accepting processor, input, or output. fields lists configurable parameters that users can pass from the pipeline config, and an empty array means the plugin has no configurable fields.

If a plugin fails to load, the first thing I'd check is whether there’s a mismatch between the command path and the actual script filename. Make sure yell_processor.py exists in the same directory as plugin.yaml.

Wiring the Plugin into a Pipeline

Create connect.yaml in the project root (one level above plugins/):

input:
  generate:
    interval: 1s
    count: 5
    mapping: |
      let events = ["user signed up from mobile-app", "payment processed for order 1042", "dashboard export requested", "session timeout on web-client", "inventory sync completed"]
      root = $events.index(counter() % 5)

pipeline:
  processors:
    - yell: {}

output:
  stdout:
    codec: lines

The generate input produces five test messages at one-second intervals, cycling through a list of event log strings that simulate upstream service traffic. The yell: {} entry under processors tells Connect to route each message through the plugin registered under the name yell. The empty braces pass no configuration to the plugin, which works because this version has no configurable fields. The stdout output prints each processed message to the terminal.

In a production pipeline, you would replace generate and stdout with actual data sources and targets such as Redpanda topics, HTTP endpoints, or database connections.

Running and Verifying the Pipeline

Start the pipeline by pointing --rpc-plugins at the plugin manifest:

rpk connect run --rpc-plugins=plugins/yell-processor/plugin.yaml connect.yaml

Connect reads the manifest, launches uv run yell_processor.py as a subprocess, establishes a gRPC connection, and registers the plugin under the name yell. The terminal output includes startup logs followed by the processed messages:

INFO Running main config from specified file    path=connect.yaml
INFO Listening for HTTP requests at: http://0.0.0.0:4195
INFO Launching a Redpanda Connect instance, use CTRL+C to close
INFO Output type stdout is now active
INFO Input type generate is now active
USER SIGNED UP FROM MOBILE-APP
PAYMENT PROCESSED FOR ORDER 1042
DASHBOARD EXPORT REQUESTED
SESSION TIMEOUT ON WEB-CLIENT
INVENTORY SYNC COMPLETED
INFO Pipeline has terminated. Shutting down the service

Each uppercase line confirms that the plugin received an event message, converted the payload to uppercase, and returned it to the pipeline. The pipeline exits automatically after five messages because of the count: 5 setting in the input config.

Making the Plugin Configurable

The basic yell processor has its behavior hardcoded. The plugin system supports runtime configuration through the fields array in plugin.yaml and the config dictionary passed to the constructor function.

To accept configuration, you replace the @redpanda_connect.processor decorator with a manual constructor pattern. The constructor is a function named processor that receives the config as a dictionary and returns an object with process() and close() methods.

Update plugins/yell-processor/yell_processor.py to this configurable version:

import asyncio
import logging

import redpanda_connect


class YellProcessor:
    def __init__(self, prefix: str, repeat_count: int):
        self.prefix = prefix
        self.repeat_count = repeat_count

    async def process(
        self, batch: redpanda_connect.MessageBatch
    ) -> list[redpanda_connect.MessageBatch]:
        results = []
        for msg in batch:
            text = msg.payload
            if isinstance(text, bytes):
                text = text.decode("utf-8")
            msg.payload = (self.prefix + str(text).upper()) * self.repeat_count
            results.append(msg)
        return [results]

    async def close(self) -> None:
        pass


def processor(config: redpanda_connect.Value) -> YellProcessor:
    if isinstance(config, dict):
        prefix = str(config.get("prefix", ""))
        repeat_count = int(config.get("repeat_count", 1))
    else:
        prefix = ""
        repeat_count = 1
    return YellProcessor(prefix, repeat_count)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(redpanda_connect.processor_main(processor))

The processor() function at the bottom replaces the decorator. Connect calls this function once at startup and passes any configuration values as a Value dictionary. The function reads prefix and repeat_count from the config with safe defaults, then returns a YellProcessor instance. The process() method operates on message batches rather than individual messages, which is the interface the dynamic plugin system uses internally to spread serialization costs across multiple messages.

Update plugin.yaml to declare the new fields:

name: yell
summary: Converts every message payload to uppercase with optional prefix and repeat.
command: ["uv", "run", "yell_processor.py"]
type: processor
fields:
  - name: prefix
    type: string
    default: ""
  - name: repeat_count
    type: int
    default: 1

Pass values for these fields in connect.yaml by replacing the empty braces with a mapping:

pipeline:
  processors:
    - yell:
        prefix: ">> "
        repeat_count: 2

Run the pipeline again with the same command. The output now includes the prefix and repeats each transformed message twice:

>> USER SIGNED UP FROM MOBILE-APP>> USER SIGNED UP FROM MOBILE-APP
>> PAYMENT PROCESSED FOR ORDER 1042>> PAYMENT PROCESSED FOR ORDER 1042
>> DASHBOARD EXPORT REQUESTED>> DASHBOARD EXPORT REQUESTED
>> SESSION TIMEOUT ON WEB-CLIENT>> SESSION TIMEOUT ON WEB-CLIENT
>> INVENTORY SYNC COMPLETED>> INVENTORY SYNC COMPLETED

Conclusion

In this tutorial, I walked you through building a custom Python processor plugin that runs inside a Redpanda Connect pipeline with full process isolation and no Go code required. I started with a minimal decorator-based processor, declared a plugin manifest, wired it into a pipeline, and then extended it to accept runtime configuration through the manifest fields.

The same pattern applies to the other two component types: inputs that pull data from external systems and outputs that push data to external targets. You can apply this approach to build custom inputs that pull from proprietary APIs, processors that call internal ML models or normalize data formats, or outputs that write to systems Redpanda Connect does not natively support. Each plugin is a standalone subprocess that you can package, version, and distribute independently. That process boundary is really the whole point, because it allows you to keep iterating in Python without ever touching the core pipeline.

The complete companion code for this tutorial is available at github.com/SystemCraftsman/redpanda-connect-dynamic-plugin-demo.

How to Configure Tracer Components in Redpanda Connect for Distributed Tracing

Aykut Bulgu — Wed, 29 Jul 2026 21:34:01 +0000

Distributed data pipelines have become a core building block of modern event-driven architectures. As these pipelines grow in complexity with multiple processors filtering, transforming, and routing messages, debugging them becomes a real challenge. When a message goes missing or a processor behaves unexpectedly, you need visibility into exactly where things broke down and how long each step took.

Distributed tracing solves this problem. Instead of relying solely on logs that tell you something happened, traces show you the full journey of every message through your pipeline. I'm going to show you how to implement that kind of tracing using Redpanda Connect's tracer component, which writes OpenTelemetry-compatible trace data directly to a Redpanda topic, so you can consume trace data the same way you consume any other event stream.

Here's what I'll walk you through:

Configure the Redpanda Connect tracer component with a Jaeger backend
Set up the OpenTelemetry Collector tracer for a backend-agnostic approach
Build a multi-step pipeline that ingests, filters, transforms, and routes application events
Explore sampling strategies for controlling trace volume in production
Visualize trace spans in the Jaeger UI to identify bottlenecks in your pipeline

Configuring Tracer Components in Redpanda Connect for Distributed Tracing

Suppose that you work for an online payment processing company that handles thousands of transaction events per second. These events come in different types such as purchase, refund, and chargeback. You've got a Redpanda Connect pipeline that ingests these events from a Redpanda input topic, filters out invalid transactions, enriches each event with additional metadata, and routes them to different output topics based on event type.

The pipeline works well, but you keep running into the same problem: when a transaction takes longer than expected or silently fails to reach its output topic, there's no easy way to pinpoint which processor caused the delay or failure. Multi-step pipelines with conditional logic are exactly where bottlenecks and silent failures hide.

The following diagram shows the high-level architecture of the pipeline with tracing enabled:

By enabling the Redpanda tracer component, every transaction event that flows through the pipeline produces a trace. Each processor in the chain generates a child span, a nested timing record showing how long that step took and whether it succeeded. That lets you see exactly how long filtering, transformation, and routing take for each event. If a refund event takes three times longer than a purchase event, the trace data will show exactly which processor is responsible.

Prerequisites

You'll need the following:

Docker and Docker Compose (Docker Engine 29 or higher)
Redpanda Connect 4.88.0 or higher installed via rpk connect install or brew install redpanda-data/tap/redpanda
rpk CLI for managing Redpanda topics and consuming trace data
Basic familiarity with Redpanda Connect's YAML pipeline configuration
Jaeger 2.17.0 or higher (runs via Docker Compose in this tutorial)

How the Redpanda Connect Tracer Component Works

Every Redpanda Connect pipeline config supports an optional top-level tracer block. When that block is present and pointing at a live backend, the runtime allocates a root span for each message the moment it enters the input. As the message flows through your processor chain, each processor appends a child span to that root. The child span captures the processor type, its execution duration, and any error status if the processor fails.

A single message trace produces a nested span hierarchy flowing from input through each processor to output:

Because trace context propagates through Kafka message headers, a Redpanda Connect pipeline sitting in the middle of a larger distributed system can both receive an upstream trace context and pass it downstream. You get end-to-end traces across multiple services, extending well beyond the pipeline boundary.

The tracer block is a sibling to input, pipeline, and output in your config file. Adding it activates trace emission without changing any pipeline behavior.

Setting Up the Demo Pipeline

I'm using a synthetic data pipeline for these examples: a generate input emitting JSON events, a bloblang processor enriching each event, a log processor confirming the transformation, and a stdout output. This setup makes it easy to observe trace structure without any external Kafka dependency.

Run the following command to create a workspace directory and navigate into it:

mkdir -p ~/redpanda-tracer-tutorial && cd ~/redpanda-tracer-tutorial

Create a file named 01-basic-pipeline.yaml with the following content:

input:
  generate:
    mapping: |
      root.user_id = uuid_v4()
      root.event = "purchase"
      root.amount = random_int(min: 1, max: 500)
    interval: 1s
    count: 20

pipeline:
  processors:
    - bloblang: |
        root = this
        root.currency = "USD"
        root.processed_at = now()
    - log:
        message: "Processed event for user ${!json(\"user_id\")}"

output:
  stdout: {}

The generate input creates 20 synthetic events at one-second intervals, each with a user_id, an event type, and a random amount. The bloblang processor enriches each event by adding currency and processed_at fields, and the log processor prints a confirmation message for each processed event.

Run the following command to confirm the pipeline works before adding tracing:

rpk connect run 01-basic-pipeline.yaml

You should see 20 JSON objects emitted to stdout. Once the base pipeline is confirmed, you're ready to add the tracer block.

Configuring the Jaeger Tracer in Redpanda Connect

Jaeger remains one of the most common local tracing backends for development. Redpanda Connect ships a native jaeger tracer type that communicates with a Jaeger agent over UDP (port 6831 by default) or with a Jaeger collector over HTTP.

Download the following docker-compose.yml file to start a local Jaeger all-in-one instance:

curl -O https://raw.githubusercontent.com/SystemCraftsman/redpanda-connect-configuring-tracer-components-demo/main/docker/docker-compose.yml

Run the following command to start Jaeger:

docker compose up jaeger

The all-in-one image exposes the Jaeger UI at http://localhost:16686, accepts HTTP collector traffic on port 14268, and accepts UDP agent traffic on port 6831.

Copy 01-basic-pipeline.yaml to a new file named 02-jaeger-tracing.yaml and add the following tracer block at the end:

tracer:
  jaeger:
    collector_url: http://localhost:14268/api/traces
    sampler_type: const
    sampler_param: 1
    tags:
      pipeline: purchase-events
      env: local

The collector_url field sends spans via HTTP to the Jaeger collector endpoint. This is the recommended approach for local Docker setups because the UDP-based agent_address option can be unreliable on the macOS Docker Desktop. sampler_type: const with sampler_param: 1 traces every message, which is appropriate for local development where you want complete visibility. The tags map attaches arbitrary key-value pairs to every span from this pipeline, which helps when you're running multiple pipelines and filtering traces in the Jaeger UI by service or environment.

Run the following command to start the traced pipeline:

rpk connect run 02-jaeger-tracing.yaml

After the 20 messages process, open http://localhost:16686, select the service named benthos (the default service name for Redpanda Connect pipelines), and click Find Traces. Each trace represents a single generated message. Click into any trace to see the root span covering the full pipeline execution time, with child spans for the bloblang processor and the log processor nested underneath.

To verify that traces are being collected, you should see 20 traces listed in the Jaeger UI. Each trace must contain a root span and two child spans corresponding to the bloblang and log processors.

The agent_address field is an alternative that sends spans via UDP to a Jaeger agent (requires port 6831):

tracer:
  jaeger:
    agent_address: localhost:6831
    sampler_type: const
    sampler_param: 1

Use agent_address in environments where a local Jaeger agent sidecar is already running and UDP connectivity is reliable. In Kubernetes environments, most teams prefer collector_url pointing at a centralized collector.

Configuring the OpenTelemetry Collector Tracer

The open_telemetry_collector tracer type sends spans via the OpenTelemetry Protocol (OTLP) over gRPC or HTTP. This approach is backend-agnostic, pointing Redpanda Connect at an OTel Collector that routes spans to Jaeger, Grafana Tempo, Honeycomb, Datadog, or any other OTLP-compatible destination.

Download the following otel-collector-config.yaml file. It accepts OTLP traffic on port 4317 (gRPC) and port 4318 (HTTP), applies a batch processor, and exports spans to the local Jaeger instance:

curl -O https://raw.githubusercontent.com/SystemCraftsman/redpanda-connect-configuring-tracer-components-demo/main/docker/otel-collector-config.yaml

The downloaded file should have the following content:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:

exporters:
  otlp_grpc/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlp_grpc/jaeger]

Run the following command to start both Jaeger and the OTel Collector together:

docker compose up

Copy 01-basic-pipeline.yaml to a new file named 03-otel-tracing.yaml and add the following tracer block at the end:

tracer:
  open_telemetry_collector:
    service: purchase-pipeline
    grpc:
      - address: localhost:4317
    tags:
      env: local
    sampling:
      enabled: false

The service field sets the service name that appears in your tracing UI. Changing it from the default benthos to something descriptive (like purchase-pipeline) makes it much easier to locate spans when you have multiple pipelines sending to the same collector.

Note that the grpc list accepts multiple endpoints, which is useful when your collector runs behind a load balancer or when you need redundancy. Each entry takes an address and an optional secure field for TLS.

Run the following command to start the pipeline:

rpk connect run 03-otel-tracing.yaml

To verify, open the Jaeger UI at http://localhost:16686 and search for the purchase-pipeline service. The span structure is identical to what you saw with the native Jaeger tracer: a root span per message with processor child spans nested underneath. You should see the traces listed under the purchase-pipeline service name instead of the default benthos.

The HTTP transport option works the same way, using port 4318 by OTLP convention. Make sure port 4318 is exposed in your docker-compose.yml (the compose file used in this tutorial already includes it):

tracer:
  open_telemetry_collector:
    service: purchase-pipeline
    http:
      - address: localhost:4318

Note that the address value does not include a protocol prefix. Redpanda Connect handles the HTTP transport internally based on the http block.

Use HTTP when your environment restricts gRPC traffic or when your OTel Collector only exposes an HTTP endpoint.

Choosing a Sampling Strategy for Redpanda Connect Traces

Sampling controls what fraction of traces gets exported. At high message volumes, exporting every span is expensive in both network overhead and backend storage. The OpenTelemetry sampling documentation covers the tradeoffs in depth.

The jaeger tracer supports two sampler types you'll use most often. With sampler_type: const and sampler_param: 1, every message produces a trace. Setting sampler_param: 0 disables tracing completely without removing the tracer block, which is useful when you want to keep the config ready but temporarily stop generating spans.

For probabilistic sampling, set sampler_type: probabilistic and give it a ratio:

tracer:
  jaeger:
    collector_url: http://localhost:14268/api/traces
    sampler_type: probabilistic
    sampler_param: 0.1

sampler_param: 0.1 samples approximately 10% of messages. That's roughly where I'd start for production pipelines doing more than 10,000 messages per second, but you can always tighten it once you see real trace volume.

The open_telemetry_collector tracer has its own sampling block:

tracer:
  open_telemetry_collector:
    service: purchase-pipeline
    grpc:
      - address: otel-collector:4317
    sampling:
      enabled: true
      ratio: 0.1

The ratio field takes a float value between 0 and 1. Wrapping it in quotes causes an unmarshal error because the parser expects a numeric type. Setting enabled: false (the default) passes all spans through to the collector and lets the collector make sampling decisions, which is the preferred pattern when you want centralized sampling policy control across multiple services via the OTel Collector's tail sampling processor.

Configuring Schema Registry for Trace Data

In the previous examples, I used generate and stdout with no external dependencies. The Redpanda tracer type, however, writes trace data to a Redpanda topic, so it requires a running Redpanda cluster with broker and schema registry endpoints accessible. You can start a local Redpanda cluster using rpk container or add a Redpanda service to your docker-compose.yml.

By default, the Redpanda tracer emits trace data in JSON format. While this works well for development, serializing trace data with Redpanda's schema registry matters for downstream consumers that need a consistent and validated schema to parse traces reliably.

To enable schema registry serialization, set format to schema-registry-json and configure the schema_registry.url in your tracer block:

tracer:
  redpanda:
    seed_brokers:
      - localhost:9092
    topic: otel-traces
    format: schema-registry-json
    schema_registry:
      url: http://localhost:8081
    service: purchase-pipeline
    tags:
      env: local

Notice that the format field supports several options: json, protobuf, schema-registry-json, and schema-registry-protobuf. The schema-registry-json option registers the trace schema with the schema registry and serializes each trace message accordingly.

You can verify the deserialized trace output by consuming the otel-traces topic with rpk:

rpk topic consume otel-traces --format json

You should see the following output with trace spans in JSON format:

{
  "trace_id": "abc123...",
  "span_id": "def456...",
  "operation_name": "bloblang",
  "service_name": "purchase-pipeline",
  "duration_ms": 12,
  "tags": {
    "env": "local"
  },
  "child_spans": [...]
}

Note: While the tracer config API is stable, the format of spans, tags, and logs is subject to change. Avoid building brittle downstream parsers that depend on specific span field names.

Adding Compression and Shutdown Delay

For production environments where trace volume can be significant, you should enable compression to reduce bandwidth and storage costs. Add compression: lz4 to your tracer configuration:

tracer:
  redpanda:
    seed_brokers:
      - localhost:9092
    topic: otel-traces
    format: schema-registry-json
    schema_registry:
      url: http://localhost:8081
    compression: lz4
    service: purchase-pipeline
    sampling:
      enabled: true
      ratio: 0.05
    tags:
      env: production

shutdown_delay: "5s"

Notice that compression supports several codecs: lz4, snappy, gzip, zstd, and none. The lz4 codec provides a good balance between compression ratio and CPU overhead.

Keep in mind that shutdown_delay is a root-level field in the pipeline configuration, not inside the tracer block. It gives the tracer time to flush remaining spans before the process exits. Without it, you may lose traces that are still buffered when the pipeline shuts down. While the shutdown delay is in effect, the HTTP metrics endpoint continues to be available for scraping and any active tracers are free to flush remaining traces.

IMPORTANT: The Redpanda Connect documentation recommends a sampling.ratio between 0.01 and 0.1 for high-throughput production environments. I've settled on starting at 0.05 (5%), as a reasonable default that balances observability with overhead.

Conclusion

In this tutorial, I showed you how to configure tracer components in Redpanda Connect for distributed tracing. You set up a multi-step pipeline, configured the native Jaeger tracer for local development, and then switched to the OpenTelemetry Collector tracer for a backend-agnostic approach. You also explored sampling strategies to control trace volume in production, configured schema registry serialization for downstream consumers, and added compression and shutdown delay for production readiness.

Because trace data lands in the otel-traces topic just like any other event, you can consume it the same way, correlating traces across pipeline instances or routing it downstream for further processing.

You can find the demo resources for this tutorial in this GitHub repository.

Winning bids in real-time: Building high-throughput AdTech RTB pipelines with Redpanda, Iceberg, and AI

Aykut Bulgu — Mon, 20 Jul 2026 21:19:41 +0000

In programmatic advertising, every millisecond counts. Real-time bidding (RTB) systems have to compete in auctions and return a bid in sub-100ms windows to win impressions, while simultaneously capturing clickstream events like views and device/campaign metadata to power audience segmentation, personalization, and model retraining. I wanted to figure out how to do both at scale, without latency spikes.

To do that, I built an RTB-style pipeline where clickstream events are ingested with Redpanda Connect, written to a data lake as Apache Iceberg tables using Redpanda Iceberg Topics, and then used to train and serve a small AI scoring model that feeds predictions back into a bidder loop.

My goal wasn't to fully implement OpenRTB end-to-end, but to build a high-throughput, low-latency pipeline skeleton that could grow into a production RTB architecture. Here's how I put it together, so you can follow along and build your own.

RTB bidding optimization

Imagine an RTB pipeline with three recurring problems:

Latency spikes during traffic bursts (you miss auctions when you exceed the response window).
High infrastructure cost from overprovisioning just to survive peak load.
Slow model refresh because historical data lives in silos and is painful to query and retrain from.

The goal is to improve the system so it can:

Ingest clickstream events (views/clicks/device/campaign) in real time.
Store the events cost-effectively as Iceberg tables for analytics and model training.
Use AI to predict impression value (CTR proxy/value score).
Stream predictions to the bidder so it can make faster and smarter decisions.

Here's the rough architecture:

To revamp the system, you'll implement the following components:

Redpanda Connect handles HTTP ingestion of bid requests and acts as the "glue", consuming requests and producing predictions.
Redpanda Iceberg Topics streams data into Iceberg tables in object storage. Redpanda writes the data in Iceberg-compatible format, making it queryable with Iceberg clients.
Spark/Jupyter (from the lab) queries Iceberg tables and produces a dataset for training.
A tiny inference service scores bid requests and publishes bid_predictions.
A simple bidder loop consumes predictions and decides how aggressively to bid.

Prerequisites

To keep this tutorial simple and repeatable, you'll use Redpanda’s official Iceberg Docker Compose lab as the base environment. It includes Redpanda, MinIO (S3-compatible storage), and Spark + Jupyter configured to query Iceberg tables.

You'll also need:

Docker and Docker Compose
rpk CLI
Python 3.11 or newer (for a tiny inference service and generators)

Bringing up the Iceberg Playground (Redpanda, MinIO, Spark/Jupyter)

To start, clone the labs repo and enter the project directory:

git clone https://github.com/redpanda-data/redpanda-labs.git && cd redpanda-labs/docker-compose/iceberg

This lab expects Redpanda >= 24.3.1. Define the required Redpanda versions:

export REDPANDA_VERSION=v25.3.2
export REDPANDA_CONSOLE_VERSION=v3.3.2

Start the environment. This provides a complete pipeline from producing data to querying Iceberg tables in Spark/Jupyter:

docker compose build && docker compose up

Next, create an rpk profile for the lab:

rpk profile create docker-compose-iceberg \
  --set=admin_api.addresses=localhost:19644 \
  --set=brokers=localhost:19092 \
  --set=schema_registry.addresses=localhost:18081

Creating the RTB topics

In this section, you'll create three topics:

clickstream (Iceberg-enabled)
bid_requests (messages to score)
bid_predictions (scored output)

For clickstream, start with key_value mode because it’s the easiest path for HTTP ingestion (you can POST JSON and treat it as the "value").

Run the following set of commands to create the topics:

rpk topic create clickstream --topic-config=redpanda.iceberg.mode=key_value ;
rpk topic create bid_requests ;
rpk topic create bid_predictions ;

Iceberg Topics are controlled via redpanda.iceberg.mode. Once you produce to an Iceberg-enabled topic, the data becomes available in object storage for Iceberg clients to consume.

Running clickstream ingestion with Redpanda Connect

In AdTech, clickstream often arrives via HTTP collectors. Redpanda Connect has an http_server input for receiving POSTed events.

Select a working directory on your local computer, create a YAML file called connect-clickstream.yaml, then paste the following content:

input:
  http_server:
    address: 0.0.0.0:4196
    path: /events
    allowed_verbs: [ POST ]

pipeline:
  processors:
    # Keep it simple: accept JSON and pass through.
    - mapping: |
        root = this

output:
  kafka_franz:
    seed_brokers: [ "redpanda:9092" ]
    topic: "clickstream"

kafka_franz is a Kafka output in Redpanda Connect.

Note: In Redpanda Cloud documentation, kafka_franz is marked deprecated in favor of unified Redpanda components. If your target environment is Cloud, consider switching outputs accordingly.

Run Redpanda Connect as a container on the same Docker network as the lab (you may need to adjust the network name depending on your Docker Compose project name):

docker run --rm -it \
  --network redpanda-labs_default \
  -p 4196:4196 \
  -v "$PWD/connect-clickstream.yaml:/connect.yaml:ro" \
  docker.redpanda.com/redpandadata/connect:latest \
  run /connect.yaml

Then, send a couple of sample clickstream events:

curl -X POST http://localhost:4196/events \
  -H 'Content-Type: application/json' \
  -d '{"user_id":101,"ad_id":55,"campaign_id":9,"event_type":"click","clicked":1,"ts":"2025-12-17T09:00:00Z"}'

curl -X POST http://localhost:4196/events \
  -H 'Content-Type: application/json' \
  -d '{"user_id":102,"ad_id":55,"campaign_id":9,"event_type":"view","clicked":0,"ts":"2025-12-17T09:00:01Z"}'

To verify the generated clickstream data, consume the clickstream data with the following command in a new terminal window:

rpk topic consume clickstream

The output should look like this:

{
  "topic": "clickstream",
  "value": "{\"user_id\":101,\"ad_id\":55,\"campaign_id\":9,\"event_type\":\"click\",\"clicked\":1,\"ts\":\"2025-12-17T09:00:00Z\"}",
  "timestamp": ...,
  "partition": 0,
  "offset": 0
}
{
  "topic": "clickstream",
  "value": "{\"user_id\":102,\"ad_id\":55,\"campaign_id\":9,\"event_type\":\"view\",\"clicked\":0,\"ts\":\"2025-12-17T09:00:01Z\"}",
  "timestamp": ...,
  "partition": 0,
  "offset": 0
}
...output omitted...

Verifying the Iceberg Writes

On your browser, open http://localhost:9001/browser and enter the credentials minio / minio123:

Open another tab and navigate to http://localhost:8888, which is the page for Jupyter Labs.

The lab notebook should guide you through querying Iceberg tables created from Redpanda topics.

You can also do a quick sanity check with Spark SQL. Run the following command to start spark-iceberg and spark-sql:

docker exec -it spark-iceberg spark-sql

On the open interface, you can query the Iceberg table.

Note: the exact catalog/table naming is shown in the lab notebook; the lab demonstrates querying lab.redpanda.<topic>.

Building a "bid value" model from Iceberg data

To create a small “bid value” model, you need to first prepare the data by parsing it from JSON and then save it as a parquet file on the disk of the Jupyter Labs container. Then you can train the model using the saved training dataset.

Preparing the data

At this point, you’ve validated the base path: HTTP -> Redpanda -> Iceberg. The next challenge is to train a model on fresh behavioral data and turn it into a low-latency scoring service.

Because you created the Iceberg topic in key_value mode, the Iceberg table stores your payload in a binary value column (plus a redpanda struct that holds record metadata).

That means the training workflow has two phases:

Extract: Cast/decode value and parse your JSON into typed columns
Train: Fit a lightweight model and save an artifact you can load in an API

Go back to Jupyter in your browser and create a new Python notebook. In the first cell, load the Iceberg table. In this lab, tables are queryable as lab.redpanda.<topic> (the lab explicitly shows querying lab.redpanda.value_schema_id_prefix, so your clickstream topic will be lab.redpanda.clickstream).

Enter the following script in the first row and run it:

clickstream_raw = spark.table("lab.redpanda.clickstream")
clickstream_raw.printSchema()

You should see a schema that includes:

A redpanda struct column (metadata)
A value column (binary payload)

To look at the raw payload, copy and paste the following script content in the second row of the notebook:

from pyspark.sql.functions import col

(clickstream_raw
  .select(col("redpanda.timestamp").alias("rp_timestamp"),
          col("value").cast("string").alias("json_value"))
  .show(5, truncate=False)
)

If your events were JSON, you should see JSON strings in json_value. If you see null or an error, it usually means one of two things:

You didn't actually publish JSON (unlikely in your curl case).
The bytes aren't UTF-8, so the cast doesn't decode cleanly.

Now you have to paste the JSON payload into typed columns. To do that, you need to define the schema you expect from your clickstream collector. In your Jupyter notebook, add the following and run it:

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

payload_schema = StructType([
    StructField("user_id", IntegerType(), True),
    StructField("ad_id", IntegerType(), True),
    StructField("campaign_id", IntegerType(), True),
    StructField("event_type", StringType(), True),
    StructField("clicked", IntegerType(), True),
    StructField("ts", StringType(), True),
])

To parse the value, add the following script and run it on your notebook:

parsed = (clickstream_raw
    .select(from_json(col("value").cast("string"), payload_schema).alias("e"))
    .select("e.*")
    .dropna(subset=["user_id", "ad_id", "campaign_id", "clicked"])
)

parsed.show(5, truncate=False)

You should be looking for clean columns like this:

user_id, ad_id, campaign_id as integers
event_type as a string
clicked as 0/1

If the clicked event is sometimes missing, you can default it (for demo) to 0:

from pyspark.sql.functions import when, lit

parsed = parsed.withColumn("clicked", when(col("clicked").isNull(), lit(0)).otherwise(col("clicked")))

In this tutorial, you won't train directly over millions of rows. You should take a sample so the notebook stays fast and repeatable:

training_df = parsed.limit(100_000)

training_path = "/home/jovyan/work/training_clickstream.parquet"
training_df.write.mode("overwrite").parquet(training_path)

training_path

At this point, you have a stable dataset on disk inside the Jupyter container.

Training the Model

To train the model, you'll use a logistic regression pipeline with one-hot encoding for event_type. This is intentionally small and "demo-grade," but it's enough to prove the pipeline.

In the same Jupyter notebook, copy and paste the following content into the next available row:

# If needed in the notebook environment:
# %pip install -q scikit-learn joblib pandas pyarrow

import pandas as pd

train_pd = pd.read_parquet(training_path)

X = train_pd[["ad_id", "campaign_id", "event_type"]]
y = train_pd["clicked"].astype(int)

from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score
import joblib

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

pre = ColumnTransformer(
    transformers=[
        ("cat", OneHotEncoder(handle_unknown="ignore"), ["event_type"]),
        ("num", "passthrough", ["ad_id", "campaign_id"]),
    ]
)

model = Pipeline(steps=[
    ("pre", pre),
    ("clf", LogisticRegression(max_iter=200))
])

model.fit(X_train, y_train)

probs = model.predict_proba(X_test)[:, 1]
print("AUC:", roc_auc_score(y_test, probs))

Then save the artifact:

artifact_path = "/home/jovyan/work/bid_value_model.joblib"
joblib.dump(model, artifact_path)
artifact_path

Now you have a model artifact that's easy to ship into a tiny inference service.

Serving the model as an inference endpoint

Now that you have a trained model artifact (bid_value_model.joblib), you need to make it usable by your online RTB path. To do so, you'll copy the model artifact out of the Jupyter container, then run the pre-created inference service.

Copy the model artifact out of the Jupyter container

You should have saved the artifact inside the Jupyter/Spark container at: /home/jovyan/work/bid_value_model.joblib.

On your host machine, open a terminal in the directory where you want to keep your inference service files and copy the model out:

docker cp spark-iceberg:/home/jovyan/work/bid_value_model.joblib ./bid_value_model.joblib

Do a quick sanity check:

ls -lh ./bid_value_model.joblib

You should see a non-zero file size (usually KBs–MBs depending on the pipeline).

WARNING: If the container name isn't spark-iceberg, run docker ps and replace it with whatever your lab uses.

Run the inference service

Run the following command on your terminal to clone the inference service demo, which uses FastAPI:

git clone https://github.com/SystemCraftsman/redpanda-rtb-iceberg-demo

Navigate into the cloned repository directory and create a Python virtual environment and activate it:

python3 -m venv .venv ;
source .venv/bin/activate

Then, install the demo project requirements by executing the following command:

pip install -r requirements.txt

Since you have all the required dependencies installed now, you can run the service:

uvicorn inference_service:app --host 0.0.0.0 --port 8000

Verifying the inference service

In a new terminal, run the following command:

curl -i http://localhost:8000/health

Your output should look like this:

{"status":"ok"}

Test the scoring with the following command:

curl -s -X POST http://localhost:8000/score \
  -H 'Content-Type: application/json' \
  -d '{"ad_id":55,"campaign_id":9,"event_type":"click"}'

Your output should look like this:

{"p_click":0.37}

If you get a 500 error, it usually means one of these things:

The model file wasn’t found (bid_value_model.joblib in the same directory, or wrong path)
A dependency mismatch (e.g., scikit-learn version issues)
Your request fields don’t match the feature names/types the pipeline expects

Scoring bid requests in-stream

In a terminal on your host, produce a couple of requests:

echo '{"request_id":"req-1","ad_id":55,"campaign_id":9,"event_type":"view"}' \
  | rpk topic produce bid_requests --format '%v\n'

echo '{"request_id":"req-2","ad_id":55,"campaign_id":9,"event_type":"click"}' \
  | rpk topic produce bid_requests --format '%v\n'

You need to consume these requests via a Redpanda Connect connector. Create a YAML file called connect-score-bids.yaml and paste the following data to configure this connector:

input:
  kafka_franz:
    seed_brokers: [ "redpanda:9092" ]
    topics: [ "bid_requests" ]
    consumer_group: "bid-scorer"

pipeline:
  processors:
    # Store the original request so we can merge it back after the HTTP call.
    - mapping: |
        meta original_request = content()

    # Call the inference API.
    # Important: because this Connect pipeline runs in Docker, "localhost" would refer to the container.
    # On macOS/Windows, use host.docker.internal to reach your host machine.
    - http:
        url: "http://host.docker.internal:8000/score"
        verb: POST
        headers:
          Content-Type: application/json

    # Merge original request + score response into one enriched event
    - mapping: |
        let req = meta("original_request").parse_json()
        let resp = this.parse_json()
        root = req
        root.p_click = resp.p_click

output:
  kafka_franz:
    seed_brokers: [ "redpanda:9092" ]
    topic: "bid_predictions"
    key: ${! json("request_id") }

Run Redpanda Connect on the same Docker network as the lab:

docker run --rm -it \
  --network redpanda-labs_default \
  -v "$PWD/connect-score-bids.yaml:/connect.yaml:ro" \
  docker.redpanda.com/redpandadata/connect:latest \
  run /connect.yaml

Leave this terminal open and the container running as it's your scoring pipeline.

To verify the bid predictions are being produced, open a new terminal and consume bid_predictions topic:

rpk topic consume bid_predictions -n 2 -f '%k %v\n'

You should see keys and enriched values similar to this:

req-1 {"request_id":"req-1","ad_id":55,"campaign_id":9,"event_type":"view","p_click":0.12}
req-2 {"request_id":"req-2","ad_id":55,"campaign_id":9,"event_type":"click","p_click":0.37}

Alternatively, you can use a bidder script written in Python to consume the data. Here's an example bidder.py:

import json
from kafka import KafkaConsumer

BROKER = "localhost:19092"
TOPIC = "bid_predictions"

consumer = KafkaConsumer(
    TOPIC,
    bootstrap_servers=[BROKER],
    auto_offset_reset="earliest",
    enable_auto_commit=True,
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
)

THRESHOLD = 0.30

print("Bidder is listening for predictions...")

for msg in consumer:
    pred = msg.value
    req_id = pred.get("request_id", "unknown")
    p = float(pred.get("p_click", 0.0))

    if p >= THRESHOLD:
        decision = "BID_AGGRESSIVE"
        bid_multiplier = 1.5
    else:
        decision = "BID_CONSERVATIVE"
        bid_multiplier = 0.5

    print(f"[{req_id}] p_click={p:.3f} => {decision} (multiplier={bid_multiplier})")

After you run bidder.py, you have a tiny “decision loop” sitting on top of your streaming pipeline. The script continuously consumes messages from the bid_predictions topic, extracts the model’s p_click score, and applies a simple threshold rule to choose a bid strategy. If p_click is above 0.30, it prints an aggressive decision as higher bid multiplier; otherwise it prints a conservative decision as lower multiplier. In a real RTB bidder, you’d plug in your actual pricing logic (budget constraints, floor prices, pacing, frequency caps), but the core idea is the same: stream in predictions, make a decision immediately, and respond fast enough to win auctions.

Conclusion

In this tutorial, I showed you how to build an RTB-style pipeline that separates high-throughput learning and low-latency decisioning. You've created a streaming backbone with Redpanda that can absorb traffic spikes, a lakehouse layer that keeps storage and historical analysis cost-effective, and an online scoring loop that can evolve independently as your features and models improve. It's not a production RTB architecture, but it's a skeleton you could grow into one.

You can find all the relevant code for the demo on Github.