Business story

You’re building a real-time order processing platform for an e-commerce company:

• Existing Oracle database with orders & customers (in the lab we’ll use PostgreSQL to simulate Oracle).


• New Couchbase (NoSQL) for fast customer session & cart data (you can simulate with Couchbase or Mongo if easier).


• 
  

  Need Confluent Kafka in the middle to stream events:

  
  
  
  
  
  • 
    order-service writes new orders to Kafka.
  
  
  • 
    payment-service, fraud-service, analytics-service consume.
  
  
  • Kafka Connect syncs data from Oracle → Kafka and Kafka → Couchbase.
  
  
  • ksqlDB / Kafka Streams does real-time aggregations (e.g., sales per minute, fraud rules).
  
  
  
  
  
  

What this lets you talk about:

• Kafka architecture & streaming design


• Confluent components: Brokers, Schema Registry, Connect, ksqlDB, Control Center
  


• SQL + NoSQL integration (Oracle/Postgres + Couchbase)


• Topics, partitions, replication, consumer groups, offsets


• Reliability, scale, monitoring, security


• How you “owned” the Confluent environment and became the escalation point

2. High-level architecture

Describe this in interviews like this:

1. Producers

• 
  

  order-service (REST API) → publishes orders events to Kafka.

  
  
  
  
  
  1. Kafka / Confluent cluster
  
  
  
  



• 3 Kafka brokers (or 1 for lab).



• 
  

  Topics:

  
  
  
  
  
  • 
    orders (3 partitions, RF=2)
  
  
  • payments
  
  
  • fraud-alerts
  
  
  • order-analytics
  
  
  
  
  
  



• 
  
  

  Schema Registry for Avro/JSON schemas.

  
  
  
  
  
  1. Stream processing
  
  
  
  
  
  



• 
  
  

  ksqlDB or Kafka Streams app:

  
  
  
  
  
  • joins orders with payments
  
  
  • flags potential fraud
  
  
  • writes fraud-alerts & order-analytics.
    
    
    
    
    1. Connectors
    
    
    
    
  
  
  
  
  
  




• JDBC Source Connector: Oracle/Postgres → topic legacy_orders.




• 
  
  

  Sink Connector: order-analytics → Couchbase collection.

  
  
  
  
  
  1. Consumers
  
  
  
  
  
  




• payment-service → consumes orders, writes to DB and publishes to payments.





• fraud-service → consumes orders+payments, publishes to fraud-alerts.




• 
  
  

  analytics-service → consumes orders & writes summary to NoSQL / analytics DB.

  
  
  
  
  
  1. Ops
  
  
  
  
  
  




• Confluent Control Center or CLI tools for monitoring.





• Basic ACLs / SSL (at least conceptually).

3. Tech stack for the lab

• 
  Platform: Docker Compose on your Mac


• 
  

  Core: Confluent Platform images:

  
  
  
  
  
  • 
    zookeeper (or KRaft mode if you want modern setup)
  
  
  • 
    kafka brokers
  
  
  • schema-registry
  
  
  • 
    ksqldb-server + ksqldb-cli
    
  
  
  • connect
  
  
  • control-center
  
  
  
  
  
  



• 
  
  

  Databases

  
  
  
  
  
  • 
    postgres (simulating Oracle)
  
  
  • 
    couchbase (or Mongo if Couchbase is too heavy)
  
  
  
  
  
  



• 
  
  

  Microservices

  
  
  
  
  
  • Language you like (Python or Node.js) for producer/consumer services.
  
  
  
  
  
  



• 
  
  

  UI

  
  
  
  
  
  • 
    kafdrop or Confluent Control Center to browse topics.
  
  
  
  
  
  

4. Step-by-step project plan

Step 1 – Bring up the Confluent stack with Docker Compose

Goal: show you can set up a Confluent environment from scratch.

• 
  

  Create docker-compose.yml with:

  
  
  
  
  
  • Zookeeper (optional if not using KRaft)
  
  
  • 1–3 Kafka brokers
  
  
  • Schema Registry
  
  
  • Connect
  
  
  • ksqlDB
  
  
  • Control Center
  
  
  • Postgres
  
  
  • Couchbase
  
  
  
  
  
  



• 
  
  

  Verify with:

  
  
  
  
  
  • docker ps
  
  
  • 
    kafka-topics CLI listing
  
  
  • Control Center UI opens in browser.
  
  
  
  
  
  

Interview mapping:

“Tell me about a Confluent environment you set up.”

“How many brokers, what replication factor, how did you run it locally for POCs?”

Step 2 – Design & create Kafka topics

Goal: talk like an architect about topics, partitions & replication.

• 
  

  Design topics:

  
  
  
  
  
  • 
    orders – 3 partitions, RF=1 or 2 (lab).
  
  
  • 
    payments – 3 partitions.
  
  
  • 
    fraud-alerts – 1 or 2 partitions.
  
  
  • 
    order-analytics – 3 partitions.
  
  
  
  



• Use kafka-topics CLI or Control Center to create them.




• Decide partition key (e.g., order_id or customer_id).

Interview mapping:

“How do you decide number of partitions?”

“How do you handle ordering?”

“What replication factor do you choose and why?”

Step 3 – Implement an order producer service

Goal: show hands-on Kafka client experience.

• 
  

  Build order-service:

  
  
  
  
  
  • Simple REST endpoint /orders that accepts an order JSON.
  
  
  • Validates & publishes to orders topic using Kafka client library.
  
  
  • Adds headers (source, correlation-id) to show best practices.
  
  
  
  


• 
  

  Demonstrate:

  
  
  
  
  
  • Fire a few orders.
  
  
  • Watch them appear in orders topic (Kafdrop or kafka-console-consumer).
  
  
  
  

Interview mapping:

“Walk me through a producer you wrote.”

“How do you handle retries, acks, idempotence?”

Step 4 – Implement consumer microservices

Goal: talk about consumer groups, scaling, offset management.

1. payment-service

• Consumes from orders (group payments-group).


• “Processes payment” (simulated) and publishes event to payments topic.

1. fraud-service

• Consumes from orders & payments (either directly or via order-payments stream later).


• Simple rule: if amount > X and country is Y → publish alert to fraud-alerts.

1. analytics-service

• Consumes from orders & writes to order_analytics table in Postgres (or pushes to order-analytics topic for Connect).

Show:

• Scaling a consumer group: run 2 instances of payment-service and watch partition assignment change.


• Show offset lag using kafka-consumer-groups --describe.

Interview mapping:

“How do consumer groups work?”

“What happens when you add/remove consumers?”

“How do you handle reprocessing / replay?”

Step 5 – Integrate with Oracle (Postgres) using Kafka Connect

Goal: show Connect and JDBC connectors.

• 
  

  In Postgres create tables:

  
  
  
  
  
  • legacy_orders
  
  
  • Insert some sample historical orders.
  
  
  
  


• 
  

  Configure JDBC Source Connector:

  
  
  
  
  
  • Source: Postgres legacy_orders.
  
  
  • Sink: legacy_orders topic.
  
  
  
  


• 
  

  Verify:

  
  
  
  
  
  • Rows from DB appear as messages in legacy_orders.
  
  
  
  

Interview mapping:

“Have you used Kafka Connect?”

“Explain how you brought data from Oracle into Kafka.”

“How do you handle schema changes?”

Step 6 – Sink analytics to Couchbase via Connect

Goal: show Kafka → NoSQL integration.

• Create Couchbase bucket/collection order_analytics.


• 
  

  Configure Sink Connector:

  
  
  
  
  
  • Source topic: order-analytics.
  
  
  • Target: Couchbase.
  
  
  
  


• 
  

  Verify:

  
  
  
  
  
  • Aggregated analytics events appear as documents in Couchbase.
  
  
  
  

Interview mapping:

“Tell us about integrating Kafka with NoSQL / Couchbase.”

“How did you configure your sink connectors?”

“How do you handle retries / DLQs?”

Step 7 – Stream processing with ksqlDB or Kafka Streams

Goal: cover Kafka Streams / kSQL & streaming architecture.

Using ksqlDB:

• 
  

  Define streams:

  
  
  
  
  
  • 
    ORDERS_STREAM on orders.
  
  
  • 
    PAYMENTS_STREAM on payments.
  
  
  
  


• 
  

  Build a joined stream:

  
  
  
  
  
  • 
    ORDERS_WITH_PAYMENTS joining by order_id.
  
  
  
  


• 
  

  Create aggregations:

  
  
  
  
  
  • Total sales per country per minute.
  
  
  • Count of “suspicious orders” flagged by simple rule.
  
  
  
  



• Output results to order-analytics topic (used by sink connector above).

Interview mapping:

“What’s your experience with Kafka Streams / kSQL?”

“How do you build a real-time pipeline end-to-end?”

“How do you design stateful vs stateless processing?”

Step 8 – Schema Registry & message evolution

Goal: talk about schemas, compatibility & governance.

• Define an Avro or JSON schema for Order and register it in Schema Registry.


• Configure producer & consumers to use that schema.


• 
  

  Demonstrate:

  
  
  
  
  
  • Add a new optional field (e.g., promo_code) → show compatibility (backward/forward).
  
  
  • Talk about what happens if you make a breaking change.
  
  
  
  

Interview mapping:

“Have you used Schema Registry?”

“How do you manage schema evolution?”

“How do you avoid breaking consumers?”

Step 9 – Reliability, monitoring & troubleshooting

Goal: show you can be the escalation point for Kafka.

Do some experiments:

• Kill one consumer instance and watch rebalance.


• Stop a connector → see lag build up, restart and recover.


• Configure producer with acks=all, retries, and discuss durability.


• 
  

  Use:

  
  
  
  
  
  • Control Center dashboards or CLI to check:
  
  
  • Consumer lag
  
  
  • Broker health
  
  
  • Topic throughput
  
  
  
  

Prepare talking points:

• How you would monitor in prod (Prometheus/Grafana, alerts on lag, disk, ISR count).


• Backup & disaster recovery strategy (snapshots, multi-AZ, mirror topics across clusters).

Interview mapping:

“How do you monitor Kafka?”

“What are common failure scenarios?”

“If a consumer is lagging badly, how do you troubleshoot?”

Step 10 – Security & access control (conceptual + minimal lab)

Goal: speak about security architecture even if lab is simple.

In lab (optional but nice):

• 
  

  Enable SASL/PLAINTEXT or at least explain how you’d:

  
  
  
  
  
  • Use SASL/SCRAM or mTLS for auth.
  
  
  • Use ACLs to restrict which services can read/write which topics.
  
  
  • Use encryption in transit (TLS) and at rest (disks).
  
  
  
  

Interview mapping:

“How do you secure a Kafka / Confluent environment?”

“How do you isolate teams & applications?”

Step 11 – Your 2-minute “project story” for interviews

Practice saying something like:

“I recently built a real-time orders platform using Confluent Kafka. The company had an existing Oracle database and was adding Couchbase as a NoSQL store.

I designed the Kafka architecture with multiple topics (orders, payments, fraud-alerts, order-analytics) and set up a Confluent stack with Schema Registry, Connect, and ksqlDB using Docker.

An order-service publishes orders to Kafka; payment-service, fraud-service, and analytics-service consume them in different consumer groups.

I used a JDBC Source Connector to stream historical data from Oracle (simulated with Postgres) into Kafka, and a sink connector to push real-time analytics into Couchbase.

On top of that I used ksqlDB to join orders and payments, detect potential fraud, and compute per-minute sales metrics.

I monitored consumer lag and broker health through Control Center, experimented with failures, and documented how to scale consumers and handle schema evolution using Schema Registry.

This project gave me end-to-end experience as the person owning the Confluent platform and integrating it with both SQL and NoSQL systems.”

✅ STEP 1 — FULL DOCKER COMPOSE (Confluent Platform)

Copy this into a file named:

docker-compose.yml

This setup includes:

• Zookeeper


• Kafka Broker


• Schema Registry


• Kafka Connect


• ksqlDB


• Control Center


• Kafdrop (UI for topics)


• Postgres


• Couchbase

✅ Here is the full working compose file

Do NOT modify anything yet. Just paste & run.

version: '3.8'



services:

  zookeeper:

    image: confluentinc/cp-zookeeper:7.5.0

    environment:

      ZOOKEEPER_CLIENT_PORT: 2181

      ZOOKEEPER_TICK_TIME: 2000

    ports:

      - "2181:2181"



  kafka:

    image: confluentinc/cp-kafka:7.5.0

    depends_on:

      - zookeeper

    ports:

      - "9092:9092"

    environment:

      KAFKA_BROKER_ID: 1

      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT

      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1



  schema-registry:

    image: confluentinc/cp-schema-registry:7.5.0

    depends_on:

      - kafka

    ports:

      - "8081:8081"

    environment:

      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092

      SCHEMA_REGISTRY_HOST_NAME: schema-registry

      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081



  kafka-connect:

    image: confluentinc/cp-kafka-connect:7.5.0

    depends_on:

      - kafka

      - schema-registry

    ports:

      - "8083:8083"

    environment:

      CONNECT_BOOTSTRAP_SERVERS: kafka:9092

      CONNECT_REST_PORT: 8083

      CONNECT_GROUP_ID: "connect-cluster"

      CONNECT_CONFIG_STORAGE_TOPIC: "connect-configs"

      CONNECT_OFFSET_STORAGE_TOPIC: "connect-offsets"

      CONNECT_STATUS_STORAGE_TOPIC: "connect-status"

      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1

      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1

      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1

      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter

      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter

      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter

      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter

      CONNECT_PLUGIN_PATH: "/usr/share/java,/etc/kafka-connect/jars"



  ksqldb-server:

    image: confluentinc/cp-ksqldb-server:7.5.0

    depends_on:

      - kafka

      - schema-registry

    ports:

      - "8088:8088"

    environment:

      KSQL_BOOTSTRAP_SERVERS: kafka:9092

      KSQL_KSQL_SCHEMA_REGISTRY_URL: http://schema-registry:8081

      KSQL_LISTENERS: http://0.0.0.0:8088



  ksqldb-cli:

    image: confluentinc/cp-ksqldb-cli:7.5.0

    depends_on:

      - ksqldb-server

    entrypoint: /bin/sh



  control-center:

    image: confluentinc/cp-enterprise-control-center:7.5.0

    depends_on:

      - kafka

      - schema-registry

      - kafka-connect

    ports:

      - "9021:9021"

    environment:

      CONTROL_CENTER_BOOTSTRAP_SERVERS: kafka:9092

      CONTROL_CENTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081



  kafdrop:

    image: obsidiandynamics/kafdrop

    depends_on:

      - kafka

    ports:

      - "9000:9000"

    environment:

      KAFKA_BROKERCONNECT: kafka:9092



  postgres:

    image: postgres:15

    restart: always

    environment:

      POSTGRES_PASSWORD: password

      POSTGRES_USER: postgres

      POSTGRES_DB: orders

    ports:

      - "5432:5432"



  couchbase:

    image: couchbase:latest

    ports:

      - "8091-8094:8091-8094"

      - "11210:11210"

✅ STEP 1 — START THE ENVIRONMENT

kafka-project/

├── docker-compose.yml


└── producer/

├── Dockerfile

├── requirements.txt

└── producer.py

Inside the folder where docker-compose.yml exists:

docker-compose up -d

Check containers:

docker ps

You should see:

• kafka


• zookeeper


• schema-registry


• connect


• ksqldb


• control-center


• kafdrop


• postgres


• couchbase

✅ STEP 2 — CREATE KAFKA TOPICS

Open a Kafka shell:

docker exec -it kafka bash

Now create topics:

1) Orders topic

kafka-topics --create --topic orders \

--bootstrap-server localhost:9092 \

--partitions 3 --replication-factor 1

2) Payments topic

kafka-topics --create --topic payments \

--bootstrap-server localhost:9092 \

--partitions 3 --replication-factor 1

3) Fraud alerts

kafka-topics --create --topic fraud-alerts \

--bootstrap-server localhost:9092 \

--partitions 1 --replication-factor 1

4) Analytics topic

kafka-topics --create --topic order-analytics \

--bootstrap-server localhost:9092 \

--partitions 3 --replication-factor 1

Verify topics

kafka-topics --list --bootstrap-server localhost:9092

You should see:

orders

payments

fraud-alerts

order-analytics

✅ STEP 2 — VISUAL CHECKS (VERY IMPORTANT)

Open the UIs in your browser:

You should verify:

1. Control Center shows Kafka is healthy


2. Kafdrop lists the four topics


3. Schema Registry endpoint returns JSON


4. ksqlDB is running

1. Folder structure

Inside your project folder (where docker-compose.yml lives), create:

2. requirements.txt

In producer/requirements.txt:

kafka-python==2.0.2

faker==30.3.0

• 
  kafka-python – Kafka client library


• 
  faker – to generate realistic fake order data (names, countries, etc.)

3. producer.py

In producer/producer.py:

import json

import os

import random

import time

from datetime import datetime



from faker import Faker

from kafka import KafkaProducer



# --- Config from environment ---

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092")

TOPIC_NAME = os.getenv("TOPIC_NAME", "orders")

SLEEP_SECONDS = float(os.getenv("SLEEP_SECONDS", "2"))



fake = Faker()



def create_producer():

    """

    Create Kafka producer with JSON serializer.

    """

    producer = KafkaProducer(

        bootstrap_servers=BOOTSTRAP_SERVERS,

        value_serializer=lambda v: json.dumps(v).encode("utf-8"),

        key_serializer=lambda k: str(k).encode("utf-8"),

        linger_ms=10,          # small batch delay

        acks="all",            # wait for all replicas (reliability)

        retries=3,             # simple retry

    )

    return producer



def generate_order(order_id: int) -> dict:

    """

    Generate a fake order event.

    """

    amount = round(random.uniform(10, 500), 2)

    country = random.choice(["US", "CA", "DE", "IN", "GB", "FR", "CN", "BR"])

    status = random.choice(["CREATED", "CONFIRMED", "CANCELLED"])



    order = {

        "order_id": order_id,

        "customer_id": fake.random_int(min=1000, max=9999),

        "amount": amount,

        "currency": "USD",

        "country": country,

        "status": status,

        "created_at": datetime.utcnow().isoformat() + "Z",

        "source": "order-service",

    }

    return order



def main():

    print(f"Connecting to Kafka at {BOOTSTRAP_SERVERS} ...")

    producer = create_producer()

    print(f"Producer created. Sending messages to topic '{TOPIC_NAME}'")



    order_id = 1

    while True:

        order = generate_order(order_id)

        key = order["order_id"]



        # send asynchronously

        future = producer.send(TOPIC_NAME, key=key, value=order)

        try:

            record_metadata = future.get(timeout=10)

            print(

                f"Sent order_id={order_id} to "

                f"topic={record_metadata.topic}, "

                f"partition={record_metadata.partition}, "

                f"offset={record_metadata.offset}"

            )

        except Exception as e:

            print(f"Error sending message: {e}")



        order_id += 1

        time.sleep(SLEEP_SECONDS)



if __name__ == "__main__":

    main()

What this does (talking points for interview):

• 
  Reliable producer: acks="all", retries=3 → waits for all replicas and retries on failure.


• 
  Partition key: uses order_id as key → all events for same order go to same partition (ordering).


• 
  JSON schema: a consistent order schema you can later register in Schema Registry.


• 
  Back-pressure friendly: small linger_ms to allow batching.

4. Dockerfile

In producer/Dockerfile:

FROM python:3.11-slim



WORKDIR /app



# Install dependencies

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt



# Copy app code

COPY producer.py .



ENV KAFKA_BOOTSTRAP_SERVERS=kafka:9092

ENV TOPIC_NAME=orders

ENV SLEEP_SECONDS=2



CMD ["python", "producer.py"]

5. Update docker-compose.yml to add the producer

At the bottom of your existing docker-compose.yml (same level as postgres, couchbase, etc.), add:

  order-producer:

    build: ./producer

    depends_on:

      - kafka

    environment:

      KAFKA_BOOTSTRAP_SERVERS: "kafka:9092"

      TOPIC_NAME: "orders"

      SLEEP_SECONDS: "2"

Note: indentation matters – make sure order-producer: is aligned with kafka:, postgres:, etc.

6. Build and start everything

From the folder with docker-compose.yml:

docker-compose up -d --build

Check containers:

docker ps

You should see order-producer running.

To see logs:

docker logs -f order-producer

You should see lines like:

Sent order_id=1 to topic=orders, partition=0, offset=0

Sent order_id=2 to topic=orders, partition=1, offset=5

...

7. Verify messages in Kafka

Option 1 – Kafdrop UI

Open:

• 
  Kafdrop → http://localhost:9000
  


• Click on topic orders
  


• You should see live messages with JSON payloads.

Option 2 – Kafka console consumer

Inside Kafka container:

docker exec -it kafka bash

kafka-console-consumer \

  --bootstrap-server localhost:9092 \

  --topic orders \

  --from-beginning

You’ll see the raw JSON messages being streamed.

8. How to talk about this in an interview

You can now say:

“I built an order-service producer in Python that publishes order events into a Kafka topic called orders. I used kafka-python with JSON serialization, configured acks=all and retries for reliability, and used order_id as the message key so all events for the same order go to the same partition to preserve ordering. The service runs as a Docker container inside the same Docker Compose network as the Confluent cluster, and continuously generates realistic orders using Faker, which we then view in Kafdrop and use as input for downstream consumers and ksqlDB.”

Great — we continue the project.

✅ STEP 4 — Folder Structure

Extend your project like this:

kafka-project/

├── docker-compose.yml

├── producer/

│   ├── Dockerfile

│   ├── producer.py

│   └── requirements.txt

└── consumers/

    ├── payment-service/

    │   ├── Dockerfile

    │   ├── requirements.txt

    │   └── payment_consumer.py

    ├── fraud-service/

    │   ├── Dockerfile

    │   ├── requirements.txt

    │   └── fraud_consumer.py

    └── analytics-service/

        ├── Dockerfile

        ├── requirements.txt

        └── analytics_consumer.py

🎯 Overview of what each service will do

1. payment-service

• Consumes from orders
  


• Processes order (simulated)


• Publishes payment result to payments topic

2. fraud-service

• Consumes from orders + payments
  


• 
  

  Simple fraud detection rule:

  
  
  
  
  
  • if amount > 300 and country not in ["US", "CA"]
    
  
  
  
  
  
  




• Publishes alert to fraud-alerts topic

3. analytics-service

• Consumes from orders
  


• 
  

  Calculates:

  
  
  
  
  
  • total sales
  
  
  • order counts
  
  
  
  
  
  




• Writes aggregated results to order-analytics topic

✅ Let’s build them one by one

🔹 payment-service

consumers/payment-service/requirements.txt

kafka-python==2.0.2

consumers/payment-service/payment_consumer.py

import json

import os

import time

from kafka import KafkaConsumer, KafkaProducer



BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092")

ORDERS_TOPIC = "orders"

PAYMENTS_TOPIC = "payments"



producer = KafkaProducer(

    bootstrap_servers=BOOTSTRAP_SERVERS,

    value_serializer=lambda v: json.dumps(v).encode("utf-8"),

    key_serializer=lambda k: str(k).encode("utf-8"),

)



consumer = KafkaConsumer(

    ORDERS_TOPIC,

    bootstrap_servers=BOOTSTRAP_SERVERS,

    group_id="payments-group",

    value_deserializer=lambda m: json.loads(m.decode("utf-8")),

    auto_offset_reset="earliest",

    enable_auto_commit=True,

)



def process_payment(order):

    # Simulated payment processing

    return {

        "order_id": order["order_id"],

        "status": "PAID",

        "amount": order["amount"],

        "country": order["country"],

        "timestamp": time.time(),

    }



print("Payment service started. Listening to 'orders' topic...")



for msg in consumer:

    order = msg.value

    print(f"[payment-service] Received order: {order}")



    payment_event = process_payment(order)



    producer.send(PAYMENTS_TOPIC, key=payment_event["order_id"], value=payment_event)

    print(f"[payment-service] Sent payment: {payment_event}")

consumers/payment-service/Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY payment_consumer.py .

CMD ["python", "payment_consumer.py"]

🔹 fraud-service

consumers/fraud-service/requirements.txt

kafka-python==2.0.2

consumers/fraud-service/fraud_consumer.py

import json

import os

from kafka import KafkaConsumer, KafkaProducer



BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092")



ORDERS_TOPIC = "orders"

PAYMENTS_TOPIC = "payments"

ALERTS_TOPIC = "fraud-alerts"



consumer = KafkaConsumer(

    ORDERS_TOPIC,

    PAYMENTS_TOPIC,

    bootstrap_servers=BOOTSTRAP_SERVERS,

    group_id="fraud-group",

    value_deserializer=lambda m: json.loads(m.decode("utf-8")),

    auto_offset_reset="earliest",

)



producer = KafkaProducer(

    bootstrap_servers=BOOTSTRAP_SERVERS,

    value_serializer=lambda v: json.dumps(v).encode("utf-8"),

)



def is_fraud(event):

    return event["amount"] > 300 and event["country"] not in ["US", "CA"]



print("Fraud service started...")



for msg in consumer:

    event = msg.value

    print(f"[fraud-service] Received event: {event}")



    if "amount" in event and "country" in event:

        if is_fraud(event):

            alert = {

                "order_id": event["order_id"],

                "reason": "High amount from risky country",

                "amount": event["amount"],

                "country": event["country"],

            }

            producer.send(ALERTS_TOPIC, value=alert)

            print(f"[fraud-service] ALERT SENT: {alert}")

consumers/fraud-service/Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY fraud_consumer.py .

CMD ["python", "fraud_consumer.py"]

🔹 analytics-service

consumers/analytics-service/requirements.txt

kafka-python==2.0.2

consumers/analytics-service/analytics_consumer.py

import json

import os

from kafka import KafkaConsumer, KafkaProducer



BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092")



ORDERS_TOPIC = "orders"

ANALYTICS_TOPIC = "order-analytics"



consumer = KafkaConsumer(

    ORDERS_TOPIC,

    bootstrap_servers=BOOTSTRAP_SERVERS,

    group_id="analytics-group",

    value_deserializer=lambda m: json.loads(m.decode("utf-8")),

    auto_offset_reset="earliest",

)



producer = KafkaProducer(

    bootstrap_servers=BOOTSTRAP_SERVERS,

    value_serializer=lambda v: json.dumps(v).encode("utf-8"),

)



total_sales = 0

order_count = 0



print("Analytics service started...")



for msg in consumer:

    order = msg.value

    amount = float(order["amount"])

    total_sales += amount

    order_count += 1



    analytics = {

        "total_sales": round(total_sales, 2),

        "order_count": order_count

    }



    producer.send(ANALYTICS_TOPIC, value=analytics)

    print(f"[analytics-service] {analytics}")

consumers/analytics-service/Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY analytics_consumer.py .

CMD ["python", "analytics_consumer.py"]

🔹 Update docker-compose.yml

At the bottom, add:

  payment-service:

    build: ./consumers/payment-service

    depends_on:

      - kafka

    environment:

      KAFKA_BOOTSTRAP_SERVERS: "kafka:9092"



  fraud-service:

    build: ./consumers/fraud-service

    depends_on:

      - kafka

    environment:

      KAFKA_BOOTSTRAP_SERVERS: "kafka:9092"



  analytics-service:

    build: ./consumers/analytics-service

    depends_on:

      - kafka

    environment:

      KAFKA_BOOTSTRAP_SERVERS: "kafka:9092"

🔥 Start Everything

From project root:

docker-compose up -d --build

View logs:

docker logs -f payment-service

docker logs -f fraud-service

docker logs -f analytics-service

🎉 You now have:

✔ Real producer → orders

✔ payment-service → payments

✔ fraud-service → fraud-alerts

✔ analytics-service → order-analytics

✔ All streaming through Confluent stack

✔ Fully Dockerized

✔ View everything in Kafdrop and Control Center

✅ STEP 5 — Kafka Connect

This step covers:

✔ JDBC Source Connector

Postgres → Kafka (legacy_orders topic)

(simulates Oracle → Kafka)

✔ Couchbase Sink Connector

Kafka (order-analytics topic) → Couchbase

(simulates Kafka → NoSQL system)

Your project will now match enterprise architectures AND answer interview questions:

• “How did you load data from Oracle into Kafka?”


• “How did you push streaming data into Couchbase?”


• “How do you manage connectors?”


• “What is SMT? What is DLQ? What is tasks.max? What is poll.interval.ms?”


• “How does Schema Registry integrate with Connect?”


• “How do you secure connectors?”


• “How do you monitor Connect?”

This step elevates your project to enterprise-grade.

🔥 Before we start

Make sure in your docker-compose:

• 
  kafka-connect service exists


• It has this config:
  

CONNECT_PLUGIN_PATH: "/usr/share/java,/etc/kafka-connect/jars"

This allows us to drop custom connector JARs into /jars.

🎯 STEP 5A — JDBC SOURCE CONNECTOR

Goal: Simulate an Oracle → Kafka pipeline using Postgres

✅ 1. Create sample table in Postgres

Open a bash shell in the Postgres container:

docker exec -it postgres bash

Enter psql:

psql -U postgres -d orders

Create table:

CREATE TABLE legacy_orders (

  order_id SERIAL PRIMARY KEY,

  customer VARCHAR(50),

  amount NUMERIC(10,2),

  country VARCHAR(10),

  created_at TIMESTAMP DEFAULT NOW()

);

Insert sample data:

INSERT INTO legacy_orders (customer, amount, country)

VALUES

('Alice', 120.50, 'US'),

('Bob', 350.00, 'BR'),

('Sam', 99.99, 'CA'),

('Julia', 499.50, 'DE');

Exit:

\q

exit

✅ 2. Install the JDBC Connector plugin

Inside your project folder:

mkdir -p kafka-connect-jars

Download the connector JAR manually:

curl -L -o kafka-connect-jars/kafka-connect-jdbc.jar \

https://packages.confluent.io/maven/io/confluent/kafka-connect-jdbc/10.7.4/kafka-connect-jdbc-10.7.4.jar

Update docker-compose.yml → inside kafka-connect service:

volumes:

  - ./kafka-connect-jars:/etc/kafka-connect/jars

Restart Kafka Connect:

docker-compose restart kafka-connect

✅ 3. Register JDBC Source Connector

Create file:

connect-jdbc-source.json

Paste:

{

  "name": "jdbc-source-legacy-orders",

  "config": {

    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",

    "tasks.max": "1",

    "connection.url": "jdbc:postgresql://postgres:5432/orders",

    "connection.user": "postgres",

    "connection.password": "password",

    "mode": "incrementing",

    "incrementing.column.name": "order_id",

    "table.whitelist": "legacy_orders",

    "topic.prefix": "legacy_",

    "poll.interval.ms": "3000"

  }

}

Send it to Kafka Connect:

curl -X POST -H "Content-Type: application/json" \

--data @connect-jdbc-source.json \

http://localhost:8083/connectors

🧪 4. Check if connector is running

List connectors:

curl localhost:8083/connectors

Check status:

curl localhost:8083/connectors/jdbc-source-legacy-orders/status

🧪 5. Verify messages appear in Kafka

Open Kafdrop → http://localhost:9000

Choose topic:

legacy_legacy_orders

You should see messages like:

{

  "order_id": 1,

  "customer": "Alice",

  "amount": 120.50,

  "country": "US",

  "created_at": "2025-11-19T10:33:00"

}

This demonstrates Oracle → Kafka ingestion.

🎉 Your JDBC Source pipeline is DONE.

🎯 STEP 5B — Couchbase SINK CONNECTOR

Goal: Write analytics data into Couchbase NoSQL bucket

✅ 1. Initialize Couchbase UI

Go to:

👉 http://localhost:8091

Initial credentials:

• Username: Administrator
  


• Password: password (set this during UI setup)

Create:

Bucket: analytics

Type: Couchbase

Quota: 256MB

Inside bucket:

• Create scope = orderscope
  


• Create collection = orderanalytics
  

Your path will be:

analytics.orderscope.orderanalytics

✅ 2. Install Couchbase Kafka Connector

Download plugin:

curl -L -o kafka-connect-jars/couchbase-kafka-connector.jar \

https://packages.couchbase.com/clients/kafka/4.1.10/couchbase-kafka-connect-couchbase-4.1.10.jar

Restart connect:

docker-compose restart kafka-connect

✅ 3. Register Couchbase Sink Connector

Create file:

connect-couchbase-sink.json

Paste:

{

  "name": "couchbase-sink-order-analytics",

  "config": {

    "connector.class": "com.couchbase.connect.kafka.CouchbaseSinkConnector",

    "tasks.max": "1",

    "couchbase.cluster.address": "couchbase",

    "couchbase.bucket": "analytics",

    "couchbase.username": "Administrator",

    "couchbase.password": "password",



    "couchbase.document.id": "${order_id}",

    "couchbase.scope": "orderscope",

    "couchbase.collection": "orderanalytics",



    "topics": "order-analytics",

    "value.converter": "org.apache.kafka.connect.json.JsonConverter",

    "value.converter.schemas.enable": "false"

  }

}

Send it:

curl -X POST -H "Content-Type: application/json" \

--data @connect-couchbase-sink.json \

http://localhost:8083/connectors

🧪 4. Verify

• Open Couchbase UI


• Navigate to:
  Buckets → analytics → orderscope → orderanalytics
  

You should see documents:

{

  "total_sales": 2345.87,

  "order_count": 17

}

These come from your analytics-service.

🎉 YOU DID IT!

Your streaming architecture is now enterprise-level:

🌐 FINAL PIPELINE DIAGRAM

       Oracle/Postgres                 Producer Service

     (legacy orders)                    (Python app)

             │                                │

             ▼                                ▼

    JDBC Source Connector  ─────────►  Kafka Topic: orders

             │                                │

             ▼                                ▼

     Kafka Topic: legacy_orders       payment-service (consumes)

                                              │

                                              ▼

                                    Kafka Topic: payments

                                              │

                                              ▼

                                     fraud-service (consumes)

                                              │

                                              ▼

                                   Kafka Topic: fraud-alerts

                                              │

                                              ▼

                                   analytics-service (consumes)

                                              │

                                              ▼

                         Kafka Topic: order-analytics

                                              │

                                              ▼

                     Couchbase Sink Connector

                                              │

                                              ▼

                          Couchbase NoSQL analytics DB

1. Open ksqlDB CLI

You already have ksqldb-server and ksqldb-cli in docker-compose.yml.

Option A – use ksqldb-cli container (recommended)

docker exec -it ksqldb-cli bash

Inside the container:

ksql http://ksqldb-server:8088

You should see a prompt like:

ksql>

If that works, you’re in.

2. Create STREAM for orders topic

Your Python producer sends JSON with fields:

• order_id


• customer_id


• amount


• currency


• country


• status


• created_at


• source

From ksql>:

CREATE STREAM orders_stream (

  order_id INT,

  customer_id INT,

  amount DOUBLE,

  currency STRING,

  country STRING,

  status STRING,

  created_at STRING,

  source STRING

) WITH (

  KAFKA_TOPIC = 'orders',

  VALUE_FORMAT = 'JSON'

);

Check:

SHOW STREAMS;

DESCRIBE EXTENDED orders_stream;

Test a few records:

SELECT * FROM orders_stream EMIT CHANGES LIMIT 5;

You should see live events from your producer.

3. Create STREAM for payments topic

Your payment-service publishes JSON like:

• order_id


• status (PAID)


• amount


• country


• timestamp

Create the stream:

CREATE STREAM payments_stream (

  order_id INT,

  status STRING,

  amount DOUBLE,

  country STRING,

  timestamp DOUBLE

) WITH (

  KAFKA_TOPIC = 'payments',

  VALUE_FORMAT = 'JSON'

);

Verify:

SELECT * FROM payments_stream EMIT CHANGES LIMIT 5;

4. Real-time analytics per country (sales)

This is perfect for interview questions like:

“How did you do real-time aggregations in Kafka?”

Create a TUMBLING WINDOW aggregation:

CREATE TABLE sales_by_country_1min AS

  SELECT

    country,

    WINDOWSTART AS window_start,

    WINDOWEND   AS window_end,

    COUNT(*)    AS order_count,

    SUM(amount) AS total_sales

  FROM orders_stream

  WINDOW TUMBLING (SIZE 1 MINUTE)

  GROUP BY country

  EMIT CHANGES;

This will create a TABLE and an internal topic.

You can query it like:

SELECT * FROM sales_by_country_1min EMIT CHANGES;

Talking point:

“We used ksqlDB tumbling windows to compute per-country order counts and total sales in 1-minute windows. That gave near real-time business metrics directly off the Kafka stream.”

5. Join orders + payments in ksqlDB

Interviews LOVE joins.

We need a TABLE on the right side of a stream–table join.

5.1 Create a TABLE of latest payment per order

CREATE TABLE payments_by_order AS

  SELECT

    order_id,

    LATEST_BY_OFFSET(status) AS payment_status,

    LATEST_BY_OFFSET(amount) AS payment_amount,

    LATEST_BY_OFFSET(country) AS payment_country

  FROM payments_stream

  GROUP BY order_id

  EMIT CHANGES;

Now you have a table keyed by order_id.

5.2 Stream–table join: orders + payments

CREATE STREAM orders_with_payments AS

  SELECT

    o.order_id        AS order_id,

    o.customer_id     AS customer_id,

    o.amount          AS order_amount,

    o.country         AS order_country,

    p.payment_status  AS payment_status,

    p.payment_amount  AS payment_amount

  FROM orders_stream o

  LEFT JOIN payments_by_order p

    ON o.order_id = p.order_id

  EMIT CHANGES;

This writes to a new topic ORDERS_WITH_PAYMENTS (default name unless you override).

Verify:

SELECT * FROM orders_with_payments EMIT CHANGES;

Talking point:

“I created a ksqlDB TABLE for payments keyed by order_id, and then did a stream–table join between orders_stream and payments_by_order. That produced a real-time enriched stream with both order and payment state that we used for analytics and fraud detection.”

6. Fraud detection stream in ksqlDB

You already have a Python fraud-service, but this lets you say:

“We also implemented fraud rules in ksqlDB for streaming detection.”

Rule:

• amount > 300


• country NOT in (‘US’, ‘CA’)

Create a new fraud alerts stream that writes to a Kafka topic (you can reuse fraud-alerts or create a new one; I’ll use existing one):

CREATE STREAM fraud_alerts_stream

  WITH (

    KAFKA_TOPIC = 'fraud-alerts',

    VALUE_FORMAT = 'JSON'

  ) AS

  SELECT

    order_id,

    customer_id,

    order_amount,

    order_country,

    payment_status,

    payment_amount,

    'HIGH_AMOUNT_RISKY_COUNTRY' AS reason

  FROM orders_with_payments

  WHERE order_amount > 300

    AND order_country NOT IN ('US', 'CA')

  EMIT CHANGES;

Now:

• 
  

  Fraud alerts get written to fraud-alerts topic from two different sources if you want:

  
  
  
  
  
  • Python fraud-service
    
  
  
  • ksqlDB fraud_alerts_stream
    
  
  
  
  
  
  

You can see them:

SELECT * FROM fraud_alerts_stream EMIT CHANGES;

Or in Kafdrop → topic fraud-alerts.

7. (Optional) ksqlDB-driven analytics topic

If you want ksqlDB (instead of Python) to write to order-analytics topic:

CREATE STREAM order_analytics_stream

  WITH (

    KAFKA_TOPIC = 'order-analytics',

    VALUE_FORMAT = 'JSON'

  ) AS

  SELECT

    country,

    COUNT(*)    AS order_count,

    SUM(amount) AS total_sales

  FROM orders_stream

  GROUP BY country

  EMIT CHANGES;

In that case you can stop the Python analytics-service and say:

“Analytics were computed entirely in ksqlDB and pushed to an order-analytics topic, which we then sink into Couchbase via a Kafka Connect sink connector.”

8. Managing queries (so you sound senior)

List running queries:

SHOW QUERIES;

You’ll see IDs like:

 Query ID       | Query Type | ...

----------------+------------

 CTAS_SALES_BY_COUNTRY_1MIN_0 | PERSISTENT | ...

 CSAS_ORDERS_WITH_PAYMENTS_1  | PERSISTENT | ...

 CSAS_FRAUD_ALERTS_STREAM_2   | PERSISTENT | ...

To stop one:

TERMINATE CTAS_SALES_BY_COUNTRY_1MIN_0;

Delete a stream/table definition:

DROP STREAM orders_with_payments DELETE TOPIC;

DROP TABLE sales_by_country_1min DELETE TOPIC;

Talking point:

“We used ksqlDB persistent queries for long-running joins and aggregations. Queries were monitored and controlled with SHOW QUERIES and TERMINATE when we needed to redeploy or change logic.”

9. How to describe this whole ksqlDB part in an interview

You can say something like:

“On top of the Kafka topics, I used ksqlDB to define streams over orders and payments. I then built tumbling-window aggregations to compute per-country sales every minute and materialized that as a ksqlDB TABLE.

Next, I created a payments_by_order TABLE and joined it with the orders_stream to produce an orders_with_payments stream, which gave us an enriched view of each order together with its payment status.

Finally, I implemented fraud detection rules in ksqlDB: for example, if an order amount was above 300 USD and the country was outside US/CA, we emitted a fraud alert into the fraud-alerts topic. Those alerts were then consumed by downstream systems or written into Couchbase via a sink connector.

All of this was running as persistent ksqlDB queries, so the system continuously processed new events in real time.”