DEV Community: Byron Hsieh

Why Data Teams Need Data Lineage: From Common Pain Points to Real-World Challenges

Byron Hsieh — Thu, 12 Mar 2026 09:24:31 +0000

Why Data Teams Need Data Lineage

From Common Pain Points to Real-World Challenges

Data Lineage has become a core component of modern data platforms. It provides transparency, traceability, and observability needed for maintainability and governance. This article first covers the common challenges faced by data teams, then explores the additional real-world difficulties encountered in highly heterogeneous SQL ecosystems — the context that motivated building an automated SQL lineage extraction tool.

1. Why Data Teams Need Data Lineage (General)

Complex dependencies make incident investigation expensive

In mature data warehouses, data flows across multiple transformation layers.

When issues occur, teams must determine:

Where does this column come from?
Which transformation introduced the issue?
What downstream tables will be impacted?

Without lineage, this requires manual code tracing — slow and error-prone.

Risky schema and logic changes without impact analysis

Schema updates or ETL refactoring require an understanding of downstream dependencies.

Without lineage:

Dashboards may break
ML pipelines may fail
Incidents appear only after deployment

Lineage makes changes predictable.

Slow onboarding and weak knowledge transfer

Most data platforms lack complete documentation.

Newcomers must learn:

Table relationships
Column semantics
End-to-end data flow

Lineage accelerates onboarding by offering a visual data map.

Data teams become a support center for business users

Common questions include:

“How is this metric calculated?”
“What is the source of this column?”
“Why does this month’s number differ?”

Without lineage, engineers manually search SQL each time.

Inefficient Data Quality (DQ) incident handling and reprocessing

After identifying a data quality issue, teams must decide:

Which downstream tables need reprocessing?
How far does the impact propagate?
What is the safe order for backfilling data?

Without lineage, reprocessing scope is often guesswork — teams either miss affected systems or waste time rerunning unnecessary jobs. Lineage provides the dependency map needed for surgical, efficient remediation.

2. Real-World Challenges: A Highly Heterogeneous SQL Ecosystem

Beyond typical issues, certain enterprise environments present extreme complexity due to decades of organic growth and diverse technology stacks. Drawing from experience in large-scale data warehouses, this section describes challenges that make automated lineage extraction particularly difficult — challenges that motivated building a specialized preprocessing and parsing pipeline.

Thousands of legacy batch jobs

In mature data warehouses, jobs accumulate over years, created by different teams with inconsistent conventions:

Highly coupled dependencies
Missing or outdated metadata
Manual dependency tracking becomes impractical at scale

Automation becomes essential.

SQL is not pure SQL: templates, embedded code, and vendor dialects

SQL rarely exists in isolation in production systems. Common patterns include:

Template-based SQL:

SELECT * FROM ${SOURCE_SCHEMA}.customer_data
WHERE batch_date = ${RUN_DATE}

SQL embedded in application code:

Python: Dynamic query construction with f-strings or string concatenation
COBOL: SQL embedded in EXEC SQL blocks
Perl: Template systems mixing procedural logic with SQL

Vendor-specific dialects:

Teradata procedural extensions (.SET, .IF, EXEC)
Oracle PL/SQL blocks
T-SQL stored procedures

Each approach complicates lineage extraction — the SQL parser must first extract the query from its host language context.

Noise from comments, multilingual content, and inconsistent naming

Legacy codebases often accumulate various forms of noise:

Comments and documentation:

Mix of inline, multi-line, and vendor-specific comment styles
Documentation in multiple languages (reflecting global or offshore teams)
Outdated comments referencing deprecated logic

Naming inconsistencies:

Column names in different languages (English, local language, or mixed)
Meaningless aliases (t1, x, temp_final_final)
Reused table names across contexts

Extracting clean lineage requires distinguishing signal from noise — comments must be removed, but not at the expense of losing vendor-specific SQL syntax.

SQL patterns too complex for off-the-shelf lineage tools

Production SQL often contains patterns that defeat generic parsers:

Computed column dependencies within the same SELECT:

SELECT 
    acct.account_id,
    CASE 
        WHEN acct.status_code = '00' THEN 'A'
        WHEN acct.status_code = '07' THEN 'C'
        ELSE 'X'
    END as derived_status,

    -- References the column defined above
    CASE 
        WHEN derived_status = 'C' THEN 0
        ELSE acct.balance
    END as effective_balance,

    -- Further chained reference
    CAST(effective_balance * rate.conversion_rate AS DECIMAL(18,2)) as converted_balance

FROM source_table acct
JOIN exchange_rates rate ON acct.currency = rate.currency

This pattern is common in financial ETL but problematic for parsers: derived_status is defined and immediately referenced in the same SELECT clause. Generic parsers fail because they expect column references to resolve to source tables, not to computed columns in the same projection.

Other challenging patterns:

Deeply nested subqueries (5+ levels) with alias renaming at each layer
Large UNION chains combining 10+ tables with overlapping column names
Dynamic table/column name resolution (metadata-driven ETL)
Vendor-specific ranking functions (QUALIFY in Teradata, TOP in T-SQL)

Each pattern requires specialized AST traversal and context tracking beyond what standard SQL parsers provide.

A preprocessing pipeline becomes required

To reliably feed SQL into lineage engines (e.g., sqllineage, sqlglot), a multi-stage preprocessing pipeline is necessary:

Cleaning and normalization:

Comment removal (preserving vendor syntax)
Template variable expansion and macro resolution
Vendor-specific syntax normalization (Teradata, Oracle, T-SQL, etc.)
Removal of multilingual content and non-standard formatting

Structure extraction:

Separating DDL metadata from DML queries
Extracting SQL from host language contexts (Python, COBOL, shell scripts)
Tokenization and syntax tree preparation

Only after this normalization can standard lineage engines produce reliable results. The preprocessing layer becomes as critical as the parser itself.

3. Conclusion

Data Lineage is not optional — it is foundational to modern data operations. In environments with large amounts of legacy SQL and heterogeneous scripting, a robust lineage pipeline is crucial for maintainability and reliability.

This context motivated the development of an automated SQL lineage extraction tool. In the next articles of this series, we'll explore:

Part 2: Getting started with Python and sqllineage for standard SQL scenarios
Part 3: System Design - A Production-Ready SQL Lineage Pipeline
- Architecture designed for real production environments with thousands of legacy SQL files
- Multi-stage processing pipeline: preprocessing → parsing → lineage extraction → export
- Happy path workflow and edge case handling strategies
- Error isolation, graceful degradation, and comprehensive logging
- Design decisions: single-threaded processing, step-based architecture, and scalability considerations
- Real-world deployment and operational characteristics

dbt + OpenLineage #1: Why dbt-ol Is a Post-Processor (Not a Plugin) — and Why It Matters

Byron Hsieh — Wed, 04 Mar 2026 14:29:19 +0000

Introduction

After exploring cloud migration patterns in my current work, I started thinking about one of the next layers: data lineage.

Legacy lineage systems built on top of SQLLineage often struggle with on-premise codebases that lack enforced standards — SQL mixed across legacy scripts and raw .sql files, runtime variables, shared temp tables, duplicate aliases. Accuracy tends to top out around 70%.

The cloud migration changed the equation. A common modern pattern is to use a Spark-based ingestion layer for data landing and dbt for transformation — and with that comes actual standards: consistent naming conventions, declarative SQL, and reproducible artifacts. That's the foundation automated lineage needs. But it also introduced a new challenge: the ingestion layer and dbt are separate tools, and getting end-to-end lineage means both need to emit events in a common format. OpenLineage is designed exactly for this. This series starts with the dbt side: given that manifest.json already captures the dependency graph, what does OpenLineage actually emit?

This series documents my hands-on exploration of openlineage-dbt. In this first post, I'll cover:

How dbt-ol works under the hood (it's not what I initially assumed)
The anatomy of an OpenLineage event: Job, Run, and Dataset
What column-level lineage actually looks like in a raw .ndjson file
Why inputs: [] is empty — and why that matters

The full project is on GitHub: openlineage-dbt

The Setup

Tech Stack

Component	Tool	Why
Transformation	dbt-core + dbt-duckdb	Lightweight, no server needed
Database	DuckDB	File-based, perfect for local learning
Lineage	openlineage-dbt	Official dbt integration
Transport	File (`.ndjson`)	Start simple, switch to Marquez later

Project Architecture

One design decision I made early on: keep the dbt project completely clean of OpenLineage configuration.

openlineage_dbt/
├── jaffle_shop/          ← pure dbt project (deployable independently)
│   ├── dbt_project.yml
│   ├── models/
│   │   ├── staging/      stg_customers, stg_orders, stg_payments
│   │   ├── intermediate/
│   │   └── marts/
│   └── seeds/            raw_customers.csv, raw_orders.csv, raw_payments.csv
├── openlineage/
│   ├── openlineage.yml   ← OL config lives here, not inside jaffle_shop/
│   └── events/           ← file transport output (.gitignored)
└── docker/               ← Marquez compose (Milestone 3)

The OpenLineage config is injected via environment variable:

export OPENLINEAGE_CONFIG=../openlineage/openlineage.yml
cd jaffle_shop
uv run dbt-ol run

Why this matters: The dbt project stays pure transformation logic; lineage config is an infrastructure concern. The same dbt project runs unchanged across environments — only the injected config differs. For this learning project, OPENLINEAGE_CONFIG is sufficient. In production, the newer OPENLINEAGE__ double-underscore env var system (e.g., OPENLINEAGE__TRANSPORT__TYPE=http) is the recommended approach — no config file needed at all.

Key Learning #1: How dbt-ol Actually Works

My first assumption was that dbt-ol intercepts dbt's execution and emits events in real time — like a plugin hooked into each model run. That turned out to be wrong.

dbt-ol is a post-processing wrapper. Here's what actually happens:

dbt-ol run
    │
    ├── 1. Run dbt normally (identical to `dbt run`)
    │       └── produces: manifest.json, run_results.json
    │
    └── 2. After dbt completes, read the artifacts
            └── parse manifest.json + run_results.json
                └── emit OpenLineage events to transport

The implications of this design:

Events are emitted after execution, not during
All lineage information comes from static artifact parsing, not runtime introspection
If dbt run fails mid-way, only the completed models get events

This is also why catalog.json matters — dbt-ol reads it (if available) to enrich output datasets with schema information (field names and DuckDB types). More on this in a later section.

The Two Artifacts

Artifact	Produced by	What dbt-ol reads from it
`manifest.json`	`dbt run` / `dbt compile`	Model graph, compiled SQL, dependencies
`run_results.json`	`dbt run`	Execution status, timing per model
`catalog.json`	`dbt docs generate`	Column names and types (output schema)

Running It

# Set config path (relative to execution directory)
export OPENLINEAGE_CONFIG=../openlineage/openlineage.yml

cd jaffle_shop
uv run dbt-ol run

One subtle gotcha: log_file_path in openlineage.yml is resolved relative to where you run dbt-ol (i.e., jaffle_shop/), not relative to the config file itself. So the path needs to account for that:

# openlineage/openlineage.yml
transport:
  type: file
  log_file_path: ../openlineage/events/events.ndjson  # relative to jaffle_shop/
  append: true

Key Learning #2: Anatomy of an OpenLineage Event

Running dbt-ol run with 3 staging models produces 8 events in events.ndjson — one JSON object per line (NDJSON format):

Line 1: parent START       → the entire dbt run
Line 2: stg_customers START
Line 3: stg_orders START
Line 4: stg_payments START
Line 5: stg_customers COMPLETE
Line 6: stg_orders COMPLETE
Line 7: stg_payments COMPLETE
Line 8: parent COMPLETE    → the entire dbt run

Every event shares the same top-level structure:

{
  "eventType": "START",
  "eventTime": "2026-02-27T14:55:51.150612+00:00",
  "job":     { ... },
  "run":     { ... },
  "inputs":  [ ... ],
  "outputs": [ ... ]
}

The Three Core Entities

Job — what the work is (static, doesn't change between runs)

"job": {
  "name": "dev.main.jaffle_shop.stg_customers",
  "namespace": "dbt",
  "facets": {
    "jobType": { "jobType": "MODEL", "integration": "DBT" },
    "sql": { "dialect": "duckdb", "query": "with source as (...) select ..." }
  }
}

Run — this specific execution instance

"run": {
  "runId": "019c9f99-59c5-75bd-...",   // UUID v7, new every execution
  "facets": {
    "parent": {
      "job": { "name": "dbt-run-jaffle_shop" },
      "run": { "runId": "019c9f99-516e-7d6a-..." }  // same across all models in this dbt run
    }
  }
}

Dataset — the data being read or written

{
  "name": "dev.main.stg_customers",
  "namespace": "duckdb://dev.duckdb"
}

Why the parent Facet Matters

Every model's run points to the same parent runId — the ID of the overall dbt run. This is what allows Marquez (or any OpenLineage backend) to group all models from a single execution together, and eventually reconstruct the full DAG from a single pipeline run.

dbt-run-jaffle_shop  (runId: 019c9f99-516e...)
    ├── stg_customers  (runId: 019c9f99-59c5-75bd..., parent → 019c9f99-516e...)
    ├── stg_orders     (runId: 019c9f99-59c5-7bf9..., parent → 019c9f99-516e...)
    └── stg_payments   (runId: 019c9f99-59c5-7a35..., parent → 019c9f99-516e...)

Key Learning #3: Column-Level Lineage in Practice

Column-level lineage (CLL) is where OpenLineage gets interesting. Without any extra configuration, dbt-ol already parses the compiled SQL and produces field-level mappings in outputs[].facets.columnLineage.

Here's stg_customers.sql:

with source as (
    select * from {{ ref('raw_customers') }}
)
select
    id                             as customer_id,
    first_name,
    last_name,
    first_name || ' ' || last_name as full_name
from source

And what OpenLineage extracts from it:

"columnLineage": {
  "fields": {
    "customer_id": {
      "inputFields": [{ "field": "id",         "name": "source" }]
    },
    "first_name": {
      "inputFields": [{ "field": "first_name", "name": "source" }]
    },
    "last_name": {
      "inputFields": [{ "field": "last_name",  "name": "source" }]
    },
    "full_name": {
      "inputFields": [
        { "field": "first_name", "name": "source" },
        { "field": "last_name",  "name": "source" }
      ]
    }
  }
}

For simple rename and concatenation, the tracking is accurate:

Output column	Source column(s)	Tracking
`customer_id`	`id`	✅ rename detected
`first_name`	`first_name`	✅ pass-through
`full_name`	`first_name` + `last_name`	✅ multi-source detected

The CTE Alias Problem

Notice "name": "source" in every inputFields entry. That's the CTE alias — not the actual table name raw_customers. OpenLineage parsed the SQL correctly, but without knowing what source resolves to, the lineage chain is broken at the CTE boundary.

raw_customers  →  source (CTE)  →  stg_customers
                  ↑
              lineage stops here (CTE alias, not resolved to actual table)

This is a SQL parser limitation — the parser sees the CTE alias source but doesn't trace it back to raw_customers. Adding catalog.json does not resolve this; "name": "source" persists in both runs. That's exactly what Milestone 2 investigates.

What's Missing: The Empty inputs[]

Every model event in this run has "inputs": []. The output dataset and its column lineage are present, but there's no record of what was read.

{
  "eventType": "COMPLETE",
  "inputs":  [],              // ← empty
  "outputs": [{ "name": "dev.main.stg_customers", "facets": { "columnLineage": {...} } }]
}

The reason: these staging models read from seed tables (raw_customers, raw_orders, raw_payments). Seeds are created by dbt seed and exist as real tables in DuckDB, but dbt-ol does not treat them as upstream lineage datasets — they don't appear in inputs[] regardless of what artifacts are present.

What catalog.json actually changes

Running dbt docs generate produces target/catalog.json, which records the actual schema (column names and types) of every table and view in the database as DuckDB sees them.

uv run dbt docs generate   # produces target/catalog.json
uv run dbt-ol run          # now reads catalog.json alongside manifest.json

With catalog.json present, the output dataset gains a SchemaDatasetFacet:

"outputs": [{
  "name": "dev.main.stg_customers",
  "facets": {
    "columnLineage": { ... },
    "schema": {
      "fields": [
        { "name": "customer_id", "type": "INTEGER" },
        { "name": "first_name",  "type": "VARCHAR" },
        { "name": "last_name",   "type": "VARCHAR" },
        { "name": "full_name",   "type": "VARCHAR" }
      ]
    }
  }
}]

inputs[] remains empty. catalog.json enriches the output side with type information — it does not resolve the upstream seed tables into input datasets.

What actually changes with catalog.json

Item	Without catalog.json	With catalog.json
`inputs[]`	`[]` empty	`[]` still empty
Output `schema` facet	absent	present (field names + DuckDB types)
`columnLineage`	present	present (unchanged)
CTE alias in `inputFields`	`"name": "source"`	`"name": "source"` (unchanged)

The gap this leaves: there's no dataset-to-dataset edge connecting raw_customers → stg_customers in the emitted events. For this learning project the seeds act as the raw data layer, but in a real pipeline where staging models read from source() references (external tables), dbt-ol would populate inputs[] — that's the scenario where catalog.json matters for input schema enrichment.

This is exactly what Milestone 2 tests: intermediate and mart models that ref staging models (not seeds) — where inputs[] is no longer empty.

Takeaways

1. dbt-ol is a post-processor, not a runtime interceptor
It reads manifest.json and run_results.json after dbt run completes. This means lineage accuracy depends entirely on what's in the artifacts — and catalog.json is what adds output schema (field names and DuckDB types) to the emitted events.

2. The Job/Run/Dataset model maps cleanly onto dbt concepts
Job = dbt model definition. Run = one execution instance. Dataset = a table or view in the database. The parent facet ties all model runs within a single dbt run together, which is what makes pipeline-level lineage possible.

3. Column-level lineage works out of the box for simple SQL
Renames, pass-throughs, and multi-column expressions are all tracked correctly. The CTE alias issue ("name": "source" instead of the actual table name) is a known parser limitation — catalog.json does not resolve it. The inputFields still reference the CTE alias even after dbt docs generate.

4. Keep lineage config separate from the dbt project
Keeping openlineage.yml outside the dbt project enforces a clean boundary: dbt handles transformation, lineage config is infrastructure. OPENLINEAGE_CONFIG works well for local learning. In production, the newer OPENLINEAGE__ double-underscore env var system (e.g., OPENLINEAGE__TRANSPORT__TYPE=http) is the recommended path — each value injected directly, no config file required. The separation habit transfers either way.

What's Next

In Milestone 2, I'll run dbt docs generate to bring catalog.json into the picture, add intermediate and mart models with more complex SQL (CTEs + JOINs + window functions), and examine how column-level lineage accuracy degrades as SQL complexity increases.

Resources

Why Multi-Agent Deployment Might Slow Down Your Redshift DDL Deployments

Byron Hsieh — Thu, 05 Feb 2026 09:16:53 +0000

Background

Our data warehouse team was scaling up deployment volumes. Based on prior experience (approximately 100 tables in 30 minutes), we knew larger deployments would take considerably longer.

When concerns about execution time arose, I noticed Azure DevOps offered multi-agent deployment capabilities. The idea seemed straightforward: distribute the workload across multiple agents to speed things up.

Before implementing, I investigated what would actually happen at the database level.

The Deployment Pattern

For each table deployment, we execute DDL and DML operations across a typical multi-layered data warehouse architecture:

Data ingestion and transformation layers — Create tables for staging, integration, and historical tracking
Presentation layer — Create views for data consumption and abstraction
ETL orchestration metadata — Update control tables that manage job dependencies and execution tracking

When scaling from ~100 tables to several hundred, this translates to:

Thousands of DDL statements (CREATE TABLE / CREATE VIEW)
Hundreds of DML operations on shared control tables

Azure DevOps Multi-Agent: Understanding the Mechanism

Azure DevOps supports parallel job execution through two primary interfaces, both achieving the same underlying behavior:

Classic Editor (Release Pipelines)

In Azure DevOps Classic Editor, you can configure parallelism via:
Agent Job > Execution Plan > Parallelism > Multi-agent

This setting duplicates the entire Agent Job across multiple agents.

YAML Pipelines

The equivalent in YAML pipelines uses the parallel strategy:

jobs:
- job: DeployTables
  strategy:
    parallel: 10  # This DUPLICATES the job 10 times
  pool:
    vmImage: 'ubuntu-latest'
  steps:
  - script: |
      # WARNING: Without work distribution logic,
      # ALL 10 agents will execute THE SAME steps!
      psql -h redshift-cluster -f deploy_all_tables.sql

Both approaches have the same fundamental behavior: They create multiple identical jobs where each agent executes the same tasks unless you explicitly write logic to distribute the work.

How to Actually Distribute Work

In Classic Editor (Release Pipelines)

You need to use predefined variables or custom logic in your tasks to determine which portion of work each agent should handle.

In YAML Pipelines

Azure DevOps provides system variables for manual work distribution:

jobs:
- job: DeployTables
  strategy:
    parallel: 10
  steps:
  - bash: |
      # Use system variables to determine which tables THIS agent should handle
      POSITION=$(System.JobPositionInPhase)  # Values: 1, 2, 3, ..., 10
      TOTAL=$(System.TotalJobsInPhase)       # Value: 10

      # Calculate this agent's workload (50 tables each)
      START=$((($POSITION - 1) * 50 + 1))
      END=$(($POSITION * 50))

      echo "Agent $POSITION deploying tables $START to $END"
      for i in $(seq $START $END); do
        psql -f "table_${i}.sql"
      done

Even with correct work distribution, DDL operations on Redshift face a fundamental limitation that makes parallelization ineffective.

The Critical Understanding

What developers often assume:

"Azure DevOps will automatically split my tables across 10 agents."

What actually happens (in both Classic Editor and YAML):

Azure DevOps creates 10 identical jobs. Each job runs the exact same steps unless you explicitly write logic to distribute the work.

Redshift Architecture: The Leader Node Constraint

This is where my investigation became crucial. To understand whether multi-agent deployment would help, I needed to understand how Redshift actually handles DDL operations.

How Redshift Handles DDL

Redshift uses a Leader Node + Compute Nodes architecture:

┌─────────────────────────────┐
│     Leader Node             │  ← All DDL executes here
│  - System Catalog Tables    │
│  - Query Planning           │
│  - Metadata Management      │
└─────────────────────────────┘
         │
    ┌────┴────┬────────┬────────┐
    ▼         ▼        ▼        ▼
[Compute] [Compute] [Compute] [Compute]  ← Only for data processing

Key finding: All DDL operations execute on the Leader Node only, regardless of how many compute nodes you have.

System Catalog Table Locking

When you execute DDL statements, Redshift must update its system catalog tables. While Redshift wraps PostgreSQL system catalogs with its own views (like PG_CLASS_INFO, PG_TABLE_DEF), the underlying PostgreSQL catalog tables are where the actual locking occurs:

Core PostgreSQL Catalog Tables (where locks occur):

pg_class — stores table/view metadata
pg_attribute — stores column definitions
pg_namespace — stores schema information
pg_depend — stores object dependencies
pg_type — stores data type definitions

Redshift Information Views (built on top):

PG_CLASS_INFO, PG_ATTRIBUTE_INFO — Redshift wrappers
PG_TABLE_DEF — Redshift-specific comprehensive view
SVV_TABLE_INFO — System view with distribution information

Critical finding: DDL operations require ACCESS EXCLUSIVE LOCKS on the underlying pg_class and pg_attribute tables, which are global singleton resources on the Leader Node. This forces all DDL statements to execute serially, regardless of:

How many compute nodes you have (4, 10, or 100)
How many Azure DevOps agents you use
How well you distribute the work

-- What would happen with 10 parallel agents executing CREATE TABLE:

Agent 1/Session 1: CREATE TABLE table_001 (...);  -- Executes (holds pg_class lock)
Agent 2/Session 2: CREATE TABLE table_021 (...);  -- Waiting for pg_class lock ⏳
Agent 3/Session 3: CREATE TABLE table_041 (...);  -- Waiting for pg_class lock ⏳
Agent 4/Session 4: CREATE TABLE table_061 (...);  -- Waiting for pg_class lock ⏳
... (Agents 5-10 all waiting) ...

-- Result: Serialized execution despite parallel agents

Why This Happens:

Redshift's architecture inherits PostgreSQL's catalog design, where DDL operations must maintain ACID properties across system metadata. The Leader Node must ensure:

Consistent metadata across all catalog tables
No concurrent modifications to object definitions
Proper dependency tracking for views and constraints

This design choice prioritizes data integrity over DDL parallelism.

Control Table Locking

Additionally, our deployment pattern includes updates to shared control tables:

DELETE FROM metadata_schema.etl_control WHERE table_name = 'example_table';
INSERT INTO metadata_schema.etl_control VALUES (...);

These operations acquire table-level exclusive locks, forcing all agents to queue for access to the same tables.

Performance Analysis: What Would Actually Happen

Based on our baseline data (~100 tables in 30 minutes) and understanding of Redshift's architecture, I projected what would happen with different approaches:

Scenario 1: Multi-Agent Without Work Distribution

strategy:
  parallel: 10
steps:
  - script: execute_all_tables.sql  # No work distribution!

Projected result:

Each of 10 agents executes all tables
Total DDL operations: N_tables × 10 agents
Database receives 10x the operations
Outcome: Severe database overload, significantly slower than baseline

Scenario 2: Multi-Agent With Correct Work Distribution

strategy:
  parallel: 10
steps:
  - bash: |
      # Correctly distribute: each agent handles subset of tables
      TABLES_PER_AGENT=$(( TOTAL_TABLES / 10 ))
      # ... execute only assigned tables

Projected result based on architecture analysis:

Expected (naive calculation):
  Sequential time: ~300 minutes (extrapolated from baseline)
  With 10 agents: 300 ÷ 10 = 30 minutes ✓

Actual (with Redshift catalog locks):
  Agent 1: [████████████████████] ~300 min (executing)
  Agent 2: [⏳⏳⏳⏳⏳⏳⏳⏳] wait → execute
  Agent 3: [⏳⏳⏳⏳⏳⏳⏳⏳] wait → execute
  ...
  Total time: ~300 minutes (serialized) + orchestration overhead

Why the overhead?

Connection Overhead: 10 simultaneous connections to Redshift (vs. 1)
Lock Contention Monitoring: Redshift Leader Node processing lock queues
Transaction Retry Logic: Agents detecting timeouts and retrying
Network Latency: Multiple agents competing for responses
Logging Overhead: 10x more connection logs and audit entries

Scenario 3: Single-Agent Sequential (Current Approach)

Projected result:

Clean serial execution
Predictable linear scaling from baseline
No orchestration overhead
Expected time: Scales linearly from baseline (~3x for 300 tables)

Performance Comparison Matrix

Operation Type	Lock Scope	Without Distribution	With Distribution	Actual Benefit
`CREATE TABLE`	`pg_class` (global)	❌ N×agents duplication	❌ Serialized	0%
`CREATE VIEW`	`pg_class`, `pg_depend`	❌ N×agents duplication	❌ Serialized	0%
Control table DML	Table-level exclusive	❌ N×agents duplication	❌ Serialized (same table)	0%
`SELECT` queries	Row-level/Data blocks	⚠️ N×agents queries	✅ True parallel	80-90%

Decision: Stay with Single-Agent

Based on this analysis, the decision was clear: do not implement multi-agent deployment.

Rationale:

No performance benefit: DDL operations would serialize at the database level regardless of CI/CD parallelism
Added complexity: Multi-agent requires work distribution logic, monitoring, and coordination
Potential risks: Connection overhead and lock contention could actually slow things down
Simpler is better: Single-agent deployment is more predictable and easier to debug

Alternative optimizations considered:

Batch Transactions

   BEGIN;
   CREATE TABLE table_001 (...);
   CREATE VIEW view_001 (...);
   -- More statements...
   COMMIT;

Split into Multiple Smaller Deployments
- Deploy in batches across multiple maintenance windows
- Reduces single-deployment duration
- More manageable rollback if issues occur
Optimize Individual DDL Statements
- Remove unnecessary IF NOT EXISTS checks
- Defer COMMENT statements to post-deployment
- Minimize column count where possible

Lessons Learned

1. Architecture Understanding Matters More Than Tooling

More CI/CD resources don't help if the database architecture doesn't support parallelism. Understanding Redshift's Leader Node serialization was more valuable than adding more Azure DevOps agents.

2. Challenge "Obvious" Solutions

"More parallelism = faster" is true for data processing (SELECT queries on compute nodes), but not for DDL operations on shared catalog tables. The "obvious" solution would have wasted implementation effort with no benefit.

3. Analyze Before Implementing

By investigating the architecture before implementation, we avoided:

Wasted engineering time building work distribution logic
Added operational complexity with no performance gain
Potential performance degradation from connection overhead

This preventive analysis saved resources and kept our deployment process simple and predictable.

4. Simple Can Be Better

Single-agent deployment with optimized batching beat complex multi-agent orchestration. Sometimes the best optimization is recognizing when complexity doesn't add value.

Key Takeaways

For Redshift DDL Deployments:

✅ Single-agent sequential deployment is optimal for DDL-heavy workloads

✅ Batch transactions reduce round-trip overhead

✅ Multiple smaller deployments manage risk better than one large deployment

✅ Optimize individual DDL statements for the most impact

❌ Multi-agent parallelism provides no benefit (serialized at Leader Node)

❌ More compute nodes don't speed up DDL (only data queries benefit)

❌ Complex orchestration adds overhead without performance gain

Universal Lesson:

Understanding system constraints prevents premature optimization.

The best solution isn't always the most sophisticated—it's the one that works with your architecture, not against it.

References

Official Documentation

Key Documentation Insights

Redshift DDL Processing: All DDL operations execute on the Leader Node, not compute nodes
Azure DevOps Multi-Agent: Both Classic Editor and YAML approaches duplicate jobs; require manual work distribution
System Catalog Locking: PostgreSQL (and Redshift) use ACCESS EXCLUSIVE locks on metadata tables during DDL operations

Kafka Producer Deep Dive: From Basics to Production-Ready Configuration

Byron Hsieh — Tue, 20 Jan 2026 15:48:11 +0000

Introduction

When I started learning Apache Kafka, the Producer seemed simple at first - just call send() and you're done, right?

Wrong.

As I dug deeper, I discovered a rich set of configurations that determine whether your messages are delivered reliably or potentially lost. Understanding these concepts transformed how I think about building data pipelines.

In this article, I'll walk you through everything I learned about Kafka Producers, from the basics to production-ready configurations.

This guide is based on the excellent course "Apache Kafka Series - Learn Apache Kafka for Beginners v3".

Producer Basics
The Sticky Partitioner
Producer Acknowledgements (acks)
Producer Retries
Idempotent Producer
Production-Ready Configuration

1. Producer Basics

Setting Up a Producer

Every Kafka Producer starts with configuration:

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());

KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

The Critical Trio: send(), flush(), close()

This is where things get interesting:

send() - Asynchronous Operation

producer.send(record);

Asynchronous: The message goes into a buffer, NOT sent immediately
If your program exits right after this, the message might never reach Kafka

flush() - Synchronous Operation

producer.flush();

Synchronous: Forces all buffered messages to be sent and blocks until complete
Useful for learning/demos, but rarely used in production (impacts performance)

close() - Cleanup

producer.close();

Shuts down the Producer and releases resources
Internally calls flush() - ensures all messages are sent
MUST be called in production to prevent resource leaks

send()          flush()         close()
  ↓               ↓               ↓
[Buffer] -----> [Send] -----> [Clean up]
(async)        (sync)         (includes flush)

Callbacks for Monitoring

Callbacks let you track message delivery:

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        log.info("Sent to topic: " + metadata.topic() +
                " partition: " + metadata.partition() +
                " offset: " + metadata.offset());
    } else {
        log.error("Error while producing", exception);
    }
});

What you can get from metadata:

topic: Which topic the message was sent to
partition: Which partition number
offset: The position of this message in the partition
timestamp: When the message was created

2. The Sticky Partitioner

When I ran my producer sending 100 messages to a topic with 3 partitions, I noticed something odd: all messages went to partition 0.

Was my code broken? Not quite. This is the Sticky Partitioner at work.

How It Works

Introduced in Kafka 2.4+, the Sticky Partitioner is the default when messages don't have a key:

Messages "stick" to one partition until a batch is full
When the batch is sent, switch to a different partition
Goal: Reduce network requests by batching more efficiently

When Does a Batch Get Sent?

A batch is sent when ANY of these conditions is met:

Trigger	Setting	Default
Size limit	`batch.size`	16384 bytes (16 KB)
Time limit	`linger.ms`	0 ms
Manual	`flush()`	-

Common Misconception

batch.size is in BYTES, not message count!

// Make batches smaller to observe partition switching
properties.setProperty("batch.size", "400");  // 400 bytes

Why Sticky is Better Than Round-Robin

Old Way (Pre-Kafka 2.4):

Each message goes to a different partition
100 messages = potentially 100 network requests

New Way (Sticky Partitioner):

Messages batch together per partition
100 messages = maybe 5 network requests
Fewer network calls = better throughput

3. Producer Acknowledgements (acks)

This is one of the most important producer configurations.

The Three Levels

Setting	Behavior	Data Loss Risk
`acks=0`	Fire and forget - don't wait for any confirmation	High
`acks=1`	Wait for leader broker only	Medium
`acks=all`	Wait for leader + all in-sync replicas	None

acks=0 (Fire and Forget)

Producer → [send] → Broker
              ↓
         (don't wait)
              ↓
         [continue]

Producer doesn't wait for any acknowledgement
Highest throughput, but data loss is possible
Use case: Metrics collection where some loss is acceptable

acks=1 (Leader Only)

Producer → [send] → Leader Broker → [commit]
                          ↓
                    [send ack back]
                          ↓
                 Replicas sync later (background)

Producer waits for leader to acknowledge
Problem: If leader fails before replication, data is lost
Was the default from Kafka 1.0 to 2.8

acks=all (Full Replication)

Producer → [send] → Leader → Replica 1 → [ack]
                      ↓           ↓
                  Replica 2 → [ack]
                      ↓
              [all acks received]
                      ↓
              [send ack to producer]

Producer waits for leader AND all in-sync replicas (ISR)
No data loss (with proper configuration)
Default since Kafka 3.0

The min.insync.replicas Setting

acks=all works together with min.insync.replicas:

Replication Factor = 3
min.insync.replicas = 2

Scenario: Only 1 broker available
Result: Producer receives NOT_ENOUGH_REPLICAS exception
        → Better to fail than to lose data!

Best practice formula:

Brokers that can fail = Replication Factor - min.insync.replicas

Example: RF=3, min.insync=2 → Can tolerate 1 broker failure

4. Producer Retries

The Retry Mechanism

When sending fails, Kafka retries automatically:

|<------------ delivery.timeout.ms (default: 2 min) ------------>|
|                                                                  |
send() → [batch] → [send] → [fail] → [wait] → [retry] → [success]
                              ↑         ↑
                         network    retry.backoff.ms
                          error      (default: 100ms)

Key Settings

Setting	Default	Description
`retries`	2147483647 (Kafka 2.1+)	Max retry attempts
`retry.backoff.ms`	100	Wait time between retries
`delivery.timeout.ms`	120000 (2 min)	Upper bound for total delivery time

Understanding delivery.timeout.ms

This is the most important timeout setting:

// Total time from send() to success or failure
properties.setProperty("delivery.timeout.ms", "120000"); // 2 minutes

What happens when timeout is reached:

producer.send(record, (metadata, exception) -> {
    if (exception instanceof TimeoutException) {
        // delivery.timeout.ms exceeded
        // Message was NOT delivered - handle it!
    }
});

The Out-of-Order Problem

With retries enabled and max.in.flight.requests.per.connection > 1:

Timeline:
1. Send batch A (messages 1-10)
2. Send batch B (messages 11-20)
3. Batch A fails (network error)
4. Batch B succeeds ← commits first!
5. Batch A retries and succeeds

Result in Kafka: [11-20, 1-10]  ← Out of order!

Solution: Use Idempotent Producer (next section)

5. Idempotent Producer

The Duplicate Problem

Without idempotence, network errors can cause duplicates:

1. Producer sends message
2. Kafka commits message to log
3. Kafka sends ack
4. Ack is LOST (network error)
5. Producer thinks it failed → retries
6. Kafka commits AGAIN → Duplicate!

The Solution

properties.setProperty("enable.idempotence", "true");

How It Works

Each producer gets a Producer ID (PID) and each message batch gets a sequence number:

Producer (PID=1) sends:
  Batch 1: seq=0 → Kafka commits
  Batch 1: seq=0 → Kafka says "already have seq=0, ignoring"

Result: No duplicates!

What Idempotence Automatically Sets

When you enable idempotence, Kafka automatically configures:

Setting	Value	Why
`acks`	all	Need confirmation from all replicas
`retries`	MAX_VALUE	Retry until delivery.timeout.ms
`max.in.flight.requests.per.connection`	5	With ordering guaranteed!

The Magic of Ordering with max.in.flight=5

You might wonder: "How can we have 5 in-flight requests AND maintain order?"

Answer: Kafka uses sequence numbers to detect out-of-order batches and rejects them, forcing the producer to retry in correct order.

Batch A (seq=0) fails, Batch B (seq=1) arrives first
Kafka: "I expected seq=0, got seq=1 - rejecting!"
Producer retries Batch A, then Batch B
Order maintained!

Default Since Kafka 3.0

Idempotent producer is enabled by default in Kafka 3.0+, but explicitly enable it for older versions or for clarity.

6. Production-Ready Configuration

Safe Producer Settings

Properties properties = new Properties();

// Connection
properties.setProperty("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());

// Reliability (Kafka 3.0+ defaults, but explicit is better)
properties.setProperty("enable.idempotence", "true");
properties.setProperty("acks", "all");
properties.setProperty("retries", String.valueOf(Integer.MAX_VALUE));
properties.setProperty("max.in.flight.requests.per.connection", "5");

// Timeouts
properties.setProperty("delivery.timeout.ms", "120000");  // 2 minutes
properties.setProperty("request.timeout.ms", "30000");    // 30 seconds

High Throughput Settings (Optional)

// Batching - wait a bit to collect more messages
properties.setProperty("linger.ms", "20");
properties.setProperty("batch.size", String.valueOf(32 * 1024)); // 32KB

// Compression - reduce network bandwidth
properties.setProperty("compression.type", "snappy");  // or "lz4", "zstd"

Configuration Cheat Sheet

┌─────────────────────────────────────────────────────────────┐
│                    PRODUCER SETTINGS                        │
├─────────────────────────────────────────────────────────────┤
│  RELIABILITY                                                │
│  ├── enable.idempotence = true     (prevent duplicates)    │
│  ├── acks = all                    (full replication)      │
│  └── retries = MAX_VALUE           (bounded by timeout)    │
├─────────────────────────────────────────────────────────────┤
│  TIMEOUTS                                                   │
│  ├── delivery.timeout.ms = 120000  (total time bound)      │
│  ├── request.timeout.ms = 30000    (per request)           │
│  └── retry.backoff.ms = 100        (between retries)       │
├─────────────────────────────────────────────────────────────┤
│  THROUGHPUT                                                 │
│  ├── linger.ms = 20                (batch collection time) │
│  ├── batch.size = 32768            (batch size in bytes)   │
│  └── compression.type = snappy     (reduce network I/O)    │
├─────────────────────────────────────────────────────────────┤
│  BROKER/TOPIC (not producer settings)                       │
│  ├── replication.factor = 3                                 │
│  └── min.insync.replicas = 2                               │
└─────────────────────────────────────────────────────────────┘

Key Takeaways

send() is asynchronous - always call close() to ensure delivery
Sticky Partitioner batches messages by partition for better throughput
acks=all + min.insync.replicas=2 = no data loss (with RF=3)
delivery.timeout.ms is the upper bound for all retries
Idempotent Producer prevents duplicates AND maintains ordering
Default since Kafka 3.0 - idempotence is on, but be explicit

Conclusion

The Kafka Producer is deceptively simple on the surface but incredibly powerful when you understand its internals:

Batching improves throughput
Acknowledgements control durability
Retries handle transient failures
Idempotence prevents duplicates

The key insight: Kafka's defaults have gotten much better over time. In Kafka 3.0+, you get idempotent, exactly-once producer semantics out of the box. But understanding why these settings matter helps you tune them for your specific use case.

This article is part of my learning journey through Apache Kafka. If you found it helpful, please give it a like and follow for more tutorials!

Course Reference: Apache Kafka Series - Learn Apache Kafka for Beginners v3

Building a Kafka Wikimedia Producer: Solving 403 Errors and Understanding Java Fundamentals

Byron Hsieh — Sat, 10 Jan 2026 10:19:56 +0000

Introduction

When I started building a Kafka producer to consume real-time data from Wikimedia, I quickly realized this wasn't just about Kafka—it was also a crash course in fundamental Java concepts and real-world troubleshooting. As someone more familiar with Python, concepts like constructors and multi-threading required some mindful learning.

What I didn't expect was hitting a 403 Forbidden error when connecting to Wikimedia's API, and a confusing import error that taught me an important lesson about Maven dependencies versus Java packages.

In this article, I'll share what I learned while building a WikimediaChangeHandler that processes real-time Wikipedia edits and sends them to Kafka, including the problems I encountered and how to solve them.

This guide is based on the excellent course "Apache Kafka Series - Learn Apache Kafka for Beginners v3", with additional troubleshooting for issues not covered in the course materials.

The Goal: Building a Real-Time Wikimedia Producer

We want to build a Kafka producer that:

Connects to Wikimedia's real-time change stream
Processes incoming events
Sends them to a Kafka topic for downstream processing

The architecture looks like this:

Wikimedia SSE Stream → EventHandler → Kafka Producer → Kafka Topic

Part 1: Understanding Constructors and Dependency Injection

The Problem: Sharing Objects Between Classes

When building the handler, I encountered a fundamental question: How do I use the KafkaProducer (created in one class) inside my WikimediaChangeHandler (another class)?

In Python, I might just import and use it, but Java has a more structured approach.

The Solution: Constructors

Here's the key insight from the instructor:

"To pass in an object from one class to another in Java, you need to implement a constructor."

Let's see this in action:

public class WikimediaChangeHandler implements EventHandler {
    KafkaProducer<String, String> kafkaProducer;
    String topic;

    // Constructor receives dependencies when the object is created
    public WikimediaChangeHandler(KafkaProducer<String, String> kafkaProducer, String topic) {
        this.kafkaProducer = kafkaProducer;
        this.topic = topic;
    }

    @Override
    public void onMessage(String event, MessageEvent messageEvent) {
        // Now we can use kafkaProducer here!
        kafkaProducer.send(new ProducerRecord<>(topic, messageEvent.getData()));
    }
}

Why Do We Need a Constructor?

The scenario:

We create a KafkaProducer in the WikimediaChangesProducer class
We need to use that same producer instance in WikimediaChangeHandler
Specifically in the onMessage method, which gets called every time Wikimedia sends us data

The constructor's role:

Acts as a "receiver" when creating the object
Stores the received objects in instance variables
Makes them available to all methods in the class

Usage:

// In WikimediaChangesProducer.java
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
String topic = "wikimedia.recentchange";

// Pass producer and topic via constructor
EventHandler eventHandler = new WikimediaChangeHandler(producer, topic);

This is Dependency Injection!

This pattern has a name: Dependency Injection (DI). Instead of creating dependencies inside a class, we "inject" them from outside.

Benefits:

✅ More flexible code
✅ Easier to test (you can inject mock objects)
✅ Clear dependencies

This Pattern Exists in Other Languages

The constructor-based dependency injection pattern isn't unique to Java:

Python:

class WikimediaChangeHandler:
    def __init__(self, kafka_producer, topic):  # Python's constructor
        self.kafka_producer = kafka_producer
        self.topic = topic

    def on_message(self, event, message_event):
        self.kafka_producer.send(self.topic, message_event.data)

JavaScript/TypeScript:

class WikimediaChangeHandler {
    constructor(kafkaProducer, topic) {  // JS constructor
        this.kafkaProducer = kafkaProducer;
        this.topic = topic;
    }

    onMessage(event, messageEvent) {
        this.kafkaProducer.send(this.topic, messageEvent.data);
    }
}

This is a universal Object-Oriented Programming concept!

Part 2: Multi-Threading and Blocking the Main Thread

The Threading Challenge

Here's something the instructor explained:

"When we call eventSource.start(), it starts a background thread to process events. If we don't block the main thread, the program will finish and all threads will stop."

The Problem

Without blocking:

eventSource.start();  // Start background thread
// Main method ends immediately
// → Main thread exits → Background thread also terminates → No data is processed!

The Solution

// Start EventSource in a background thread
eventSource.start();

// Block the main program for 10 minutes to let the background thread work
TimeUnit.MINUTES.sleep(10);

Why this works:

When the main thread exits, the JVM shuts down all background threads
By using Thread.sleep(), we keep the main thread alive
This gives the background thread time to receive and process Wikimedia data

Real-World Analogy

Think of it like running a restaurant:

❌ Without blocking: You hire a chef, then immediately close the restaurant—the chef can't cook anything.

✅ With blocking: You hire a chef and keep the restaurant open for 10 minutes—the chef can do their work.

Part 3: Troubleshooting - The 403 Forbidden Error

The Problem: Course Code is Outdated

When I ran the program with the exact code from the course, I immediately hit an error:

[okhttp-eventsource-events-[]-0] ERROR WikimediaChangeHandler - Error in Stream Reading
com.launchdarkly.eventsource.UnsuccessfulResponseException:
Unsuccessful response code received from stream: 403

The program kept trying to reconnect but failed every time with HTTP 403 Forbidden.

Why This Happens

Wikimedia's API Policy Changed: Wikimedia now requires all clients connecting to their streaming API to include a User-Agent HTTP header. Without it, the server rejects the connection.

This is similar to a security guard at a building asking you to sign in. If you refuse to identify yourself, you're not allowed to enter.

The course materials haven't been updated to reflect this API policy change, so the code provided will fail.

The Solution: Add User-Agent Header

We need to modify the EventSource creation to include HTTP headers:

Original code (from course - doesn't work):

EventSource.Builder builder = new EventSource.Builder(eventHandler, URI.create(url));
EventSource eventSource = builder.build();

Fixed code:

// Add User-Agent header required by Wikimedia
import okhttp3.Headers;

Headers headers = new Headers.Builder()
    .add("User-Agent", "KafkaLearningProject/1.0")
    .build();

EventSource.Builder builder = new EventSource.Builder(eventHandler, URI.create(url))
    .headers(headers);  // Add headers here
EventSource eventSource = builder.build();

What's happening:

We create a Headers object using the Builder pattern (from OkHttp library)
Add a User-Agent header with our application identifier
Pass the headers to EventSource using .headers(headers)

User-Agent format:

Standard format: ApplicationName/Version
With contact info: "KafkaLearningProject/1.0 (your.email@example.com)"

Part 4: Troubleshooting - Maven Coordinates vs Java Packages

The Import Error Mystery

When I tried to import the Headers class, I encountered another confusing error:

error: package com.squareup.okhttp3 does not exist
import com.squareup.okhttp3.Headers;
                           ^

I thought: "But I have the dependency in build.gradle! Let me check..."

dependencies {
    implementation 'com.squareup.okhttp3:okhttp:4.9.3'  // It's here!
}

The dependency was there, and Gradle successfully downloaded it. So why couldn't Java find the class?

The Key Insight: Maven Coordinates ≠ Java Packages

This taught me an important lesson about the difference between Maven coordinates and Java package names.

Maven Coordinates (for dependency management):

groupId:artifactId:version
↓       ↓          ↓
com.squareup.okhttp3:okhttp:4.9.3

Java Package (for import statements):

import okhttp3.Headers;  // Not com.squareup.okhttp3!

Why Are They Different?

Aspect	Maven Coordinates	Java Package
Purpose	Uniquely identify library in repository	Organize code in the codebase
Example	`com.squareup.okhttp3:okhttp`	`okhttp3`
Usage	`build.gradle` dependencies	`import` statements
Reason	Avoid naming conflicts across organizations	Keep code concise and clean

Real-world examples:

Library	Maven Dependency	Java Import
OkHttp	`com.squareup.okhttp3:okhttp`	`import okhttp3.Headers;`
Gson	`com.google.code.gson:gson`	`import com.google.gson.Gson;`
Jackson	`com.fasterxml.jackson.core:jackson-databind`	`import com.fasterxml.jackson.databind.ObjectMapper;`

How to Find the Correct Package Name

Method 1: Check Official Documentation (Fastest)

// OkHttp docs clearly show:
import okhttp3.*;

Method 2: Use IDE Auto-Complete (Most Reliable)

Type: Headers headers = new Headers...
IDE shows error, press Cmd + . (Mac) or Ctrl + . (Windows)
Select "Import 'Headers' (okhttp3)"
IDE adds correct import automatically

Method 3: Inspect the JAR File (When in doubt)

jar tf okhttp-4.9.3.jar | grep Headers
# Output:
# okhttp3/Headers.class          ← The actual package!
# okhttp3/Headers$Builder.class

The Fix

Wrong import:

import com.squareup.okhttp3.Headers;  // ❌ Compilation error!

Correct import:

import okhttp3.Headers;  // ✅ Works!

Lesson Learned

Never assume the package name from the Maven coordinates!

Maven coordinates use organization namespacing (com.squareup.okhttp3)
Java packages prioritize simplicity and backwards compatibility (okhttp3)
Always verify the actual package name before importing

Complete Code Overview

Here's the complete, working code with all fixes applied:

WikimediaChangeHandler.java

package io.conduktor.demos.kafka.wikimedia;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.launchdarkly.eventsource.EventHandler;
import com.launchdarkly.eventsource.MessageEvent;

public class WikimediaChangeHandler implements EventHandler {
    private final KafkaProducer<String, String> kafkaProducer;
    private final String topic;
    private final Logger log = LoggerFactory.getLogger(WikimediaChangeHandler.class);

    /*
     * Why do we need a constructor?
     * 1. We created a KafkaProducer in the Producer class
     * 2. We need to use it in this WikimediaChangeHandler class
     * 3. Specifically in the onMessage method to send data to Kafka
     *
     * In Java, to pass objects between classes, we use constructors
     * The constructor receives parameters when creating the object and stores them for later use
     */
    public WikimediaChangeHandler(KafkaProducer<String, String> kafkaProducer, String topic) {
        this.kafkaProducer = kafkaProducer;
        this.topic = topic;
    }

    @Override
    public void onMessage(String event, MessageEvent messageEvent) {
        // When we receive a message from the stream, send it to Kafka
        log.info(messageEvent.getData());
        kafkaProducer.send(new ProducerRecord<>(topic, messageEvent.getData()));
    }

    @Override
    public void onOpen() {
        // Connection opened
    }

    @Override
    public void onClosed() {
        // Close the producer when stream closes
        kafkaProducer.close();
    }

    @Override
    public void onComment(String comment) {
        // Handle comments (not used in this case)
    }

    @Override
    public void onError(Throwable t) {
        log.error("Error in Stream Reading", t);
    }
}

WikimediaChangesProducer.java

package io.conduktor.demos.kafka.wikimedia;

import com.launchdarkly.eventsource.EventHandler;
import com.launchdarkly.eventsource.EventSource;
import okhttp3.Headers;  // Correct import!
import java.net.URI;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;

public class WikimediaChangesProducer {

    public static void main(String[] args) throws InterruptedException {
        String bootstrapServers = "127.0.0.1:9092";

        // Create and set Producer properties
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());

        // Create the Producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        String topic = "wikimedia.recentchange";

        // Create EventHandler with dependency injection
        EventHandler eventHandler = new WikimediaChangeHandler(producer, topic);

        String url = "https://stream.wikimedia.org/v2/stream/recentchange";

        // Wikimedia requires a User-Agent header to identify the client
        Headers headers = new Headers.Builder()
            .add("User-Agent", "KafkaLearningProject/1.0")
            .build();

        // Create EventSource with headers
        EventSource.Builder builder = new EventSource.Builder(eventHandler, URI.create(url))
            .headers(headers);
        EventSource eventSource = builder.build();

        // Start EventSource, which continuously receives real-time data from Wikimedia in a background thread
        eventSource.start();

        // Block the main program to prevent the main thread from terminating and stopping the background thread
        // Let the program run for 10 minutes to give the background thread enough time to receive and process data
        TimeUnit.MINUTES.sleep(10);
    }
}

Key Takeaways

1. Constructors Enable Dependency Injection

Constructors receive objects when creating instances
This is how we share objects between classes in Java
It's a universal OOP pattern, not unique to Java

2. Multi-Threading Requires Careful Management

Background threads die when the main thread exits
Use Thread.sleep() to keep the main thread alive
In production, use shutdown hooks or latches for graceful shutdown

3. APIs Change - Always Test Course Code

Course materials can become outdated as APIs evolve
Wikimedia now requires User-Agent headers (wasn't required before)
Always check API documentation if you encounter 403/401 errors

4. Maven Coordinates ≠ Java Packages

Don't assume package names from build.gradle dependencies
Maven uses organizational namespacing (com.squareup.okhttp3)
Java packages prioritize simplicity (okhttp3)
Use IDE auto-complete or check documentation for correct imports

5. Builder Pattern Provides Flexibility

Builder pattern allows optional configuration
.headers() wasn't needed before, but easy to add when required
Two-step process: configure → build

Conclusion

Building this Kafka Wikimedia producer taught me fundamental Java concepts and valuable troubleshooting skills:

Dependency Injection through constructors - how to share objects between classes
Multi-threading management - why we need to block the main thread
API troubleshooting - dealing with 403 errors and outdated course materials
Dependency management - understanding Maven coordinates vs Java packages

The most important lesson? When following online courses, the code might not work as-is. APIs change, libraries update, and policies evolve. Being able to troubleshoot these issues is just as valuable as learning the core concepts.

These challenges made me a better developer by forcing me to:

Read error messages carefully
Understand the tools I'm using (Gradle, Maven, HTTP)
Check library documentation
Learn the difference between dependency management and code organization

Troubleshooting Checklist

If you encounter issues while building this project:

403 Forbidden Error:

✅ Add User-Agent header to EventSource
✅ Ensure headers are properly built with OkHttp's Headers.Builder
✅ Check Wikimedia's API documentation for current requirements

Import Errors:

✅ Don't assume package name from Maven coordinates
✅ Use import okhttp3.Headers; not import com.squareup.okhttp3.Headers;
✅ Let IDE auto-complete suggest the correct import
✅ Run gradle --refresh-dependencies if needed

Connection Errors:

✅ Ensure Kafka broker is running on port 9092
✅ Create topic: wikimedia.recentchange
✅ Check network connectivity

This article is part of my learning journey through Apache Kafka. If you found it helpful, please give it a like and follow for more Kafka tutorials!

Course Reference: Apache Kafka Series - Learn Apache Kafka for Beginners v3

Kafka Consumer Rebalancing: From Stop-the-World to Cooperative Protocol

Byron Hsieh — Sun, 04 Jan 2026 14:10:34 +0000

Introduction

When learning about Kafka consumer groups, I discovered an important concept called rebalancing - the process where Kafka redistributes partitions among consumers when the group changes.

What I learned is that when a consumer joins or leaves a group, all consumers pause briefly during the partition reassignment. This behavior isn't a bug - it's a deliberate design choice, and Kafka actually offers two different strategies for handling it.

In this article, I'll explain the two rebalancing strategies (Eager and Cooperative), show real logs from my experiments, and discuss the trade-offs to help you choose the right strategy for your use case.

This guide is based on the excellent course "Apache Kafka Series - Learn Apache Kafka for Beginners v3".

Two Rebalancing Strategies

Kafka offers two fundamentally different approaches to rebalancing. Each has trade-offs depending on your use case.

Strategy 1: Eager Rebalance (Stop-the-World)

How It Works

Trigger event occurs (consumer joins/leaves)
ALL consumers stop consuming (stop-the-world event)
All consumers give up their partition assignments
Kafka reassigns partitions to all consumers
Consumers resume processing with new assignments

Characteristics

Simplicity:

✅ Single-step process - easy to reason about
✅ Clean state transitions - all consumers synchronized
✅ Simpler implementation and debugging

Trade-offs:

⚠️ Complete pause in processing during rebalance
⚠️ All consumers affected, even if their partitions don't change
⚠️ Local state/caches must be rebuilt after reassignment

When to Use

Eager rebalancing works well when:

Small consumer groups (2-5 consumers)
Infrequent scaling events
Rebalance duration is acceptable (typically seconds)
Simplicity is valued over minimal disruption
Consumers are stateless or have minimal state

When Does Rebalancing Trigger?

Consumer joins or leaves the group
Consumer crashes or becomes unresponsive
session.timeout.ms expires

Strategy 2: Cooperative Rebalance (Incremental)

How It Works

Trigger event occurs
Kafka identifies only the partitions that need to move
Only affected consumers pause those specific partitions
Other partitions continue processing uninterrupted
May take multiple iterations to reach stable state

Characteristics

Minimal Disruption:

✅ Only revoked partitions pause
✅ Non-affected partitions keep consuming
✅ Sticky assignment - partitions stay with consumers when possible
✅ Lower latency impact

Trade-offs:

⚠️ More complex - multiple rebalance steps
⚠️ Harder to debug (multi-phase process)
⚠️ Requires all consumers to support the protocol

When to Use

Cooperative rebalancing is beneficial when:

Large consumer groups (10+ consumers)
Frequent scaling events (auto-scaling, deployments)
Stateful consumers with large local caches
Processing interruption is costly
High-throughput systems where pauses impact SLAs

Example Scenario

Setup: 3 partitions, 2 consumers, then 1 new consumer joins

Eager Rebalance:

Before: Consumer 1: [P0, P1]    Consumer 2: [P2]
        ↓ [ALL STOP]
After:  Consumer 1: [P0]    Consumer 2: [P1]    Consumer 3: [P2]

All consumers stopped, all partitions reassigned.

Cooperative Rebalance:

Before: Consumer 1: [P0, P1]    Consumer 2: [P2]
        ↓ [Only P1 pauses]
After:  Consumer 1: [P0]    Consumer 2: [P2]    Consumer 3: [P1]

Only partition P1 moved, others kept consuming.

Partition Assignment Strategies

Kafka provides multiple assignment strategies via the partition.assignment.strategy config.

Eager Strategies (Stop-the-World)

1. RangeAssignor

Assigns partitions on per-topic basis
Can lead to imbalanced assignments
Old default strategy

2. RoundRobin

Distributes partitions evenly across consumers
All consumers have ±1 the same number of partitions
Better balance than RangeAssignor

3. StickyAssignor

Balanced like RoundRobin initially
Minimizes partition movements during rebalance
Still causes stop-the-world event

Cooperative Strategy

4. CooperativeStickyAssignor

Uses cooperative rebalancing protocol
Minimizes partition movements
Consumers keep processing non-moved partitions
Preferred for large-scale, stateful systems

Default Configuration

Kafka 3.0+ Default

partition.assignment.strategy = [RangeAssignor, CooperativeStickyAssignor]

Why both?

Provides backward compatibility
Allows gradual migration from eager to cooperative
Group coordinator picks the first strategy supported by all members

Other Components

Kafka Connect: Cooperative rebalance enabled by default
Kafka Streams: Uses StreamsPartitionAssignor (cooperative) by default

Implementing Cooperative Rebalancing

Configuration

Add this property to your consumer:

properties.setProperty("partition.assignment.strategy",
    CooperativeStickyAssignor.class.getName());

Before:

partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor,
                                  org.apache.kafka.clients.consumer.CooperativeStickyAssignor]

After:

partition.assignment.strategy = [org.apache.kafka.clients.consumer.CooperativeStickyAssignor]

Real Logs: Observing Cooperative Rebalance

I ran consumers with CooperativeStickyAssignor enabled and captured the logs during different scaling events.

Scenario 1: Single Consumer Starts

First consumer joins and gets all 3 partitions:

[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
  [Consumer clientId=consumer-my-java-application-1, groupId=my-java-application]
  Updating assignment with
        Assigned partitions:                       [demo_java-0, demo_java-1, demo_java-2]
        Current owned partitions:                  []
        Added partitions (assigned - owned):       [demo_java-0, demo_java-1, demo_java-2]
        Revoked partitions (owned - assigned):     []

State: Consumer 1 owns partitions 0, 1, 2

Scenario 2: Second Consumer Joins (Scale Up)

A new consumer joins the group. Watch how only partition 2 is revoked:

Consumer 1 - Revokes partition 2

[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
  [Consumer clientId=consumer-my-java-application-1, groupId=my-java-application]
  Updating assignment with
        Assigned partitions:                       [demo_java-0, demo_java-1]
        Current owned partitions:                  [demo_java-0, demo_java-1, demo_java-2]
        Added partitions (assigned - owned):       []
        Revoked partitions (owned - assigned):     [demo_java-2]  ← Only this one!

Key insight: Consumer 1 continues processing partitions 0 and 1 during this operation.

Consumer 1 - Assignment stabilizes

[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
  [Consumer clientId=consumer-my-java-application-1, groupId=my-java-application]
  Updating assignment with
        Assigned partitions:                       [demo_java-0, demo_java-1]
        Current owned partitions:                  [demo_java-0, demo_java-1]
        Added partitions (assigned - owned):       []
        Revoked partitions (owned - assigned):     []

Consumer 2 - Receives the revoked partition

[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
  [Consumer clientId=consumer-my-java-application-1, groupId=my-java-application]
  Updating assignment with
        Assigned partitions:                       [demo_java-2]
        Current owned partitions:                  []
        Added partitions (assigned - owned):       [demo_java-2]
        Revoked partitions (owned - assigned):     []

Result:

Consumer 1: partitions 0, 1 (kept processing throughout)
Consumer 2: partition 2 (received smoothly)
Only 1 partition moved!

Scenario 3: Consumer Leaves (Scale Down)

Consumer 2 shuts down, Consumer 1 picks up the orphaned partition:

[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
  [Consumer clientId=consumer-my-java-application-1, groupId=my-java-application]
  Updating assignment with
        Assigned partitions:                       [demo_java-0, demo_java-1, demo_java-2]
        Current owned partitions:                  [demo_java-0, demo_java-1]
        Added partitions (assigned - owned):       [demo_java-2]  ← Picked up orphan
        Revoked partitions (owned - assigned):     []

Result: Consumer 1 seamlessly adds partition 2 while continuing to process 0 and 1.

Choosing Between Strategies

Aspect	Eager (RangeAssignor)	Cooperative (CooperativeStickyAssignor)
Partition Revocation	ALL partitions	Only affected partitions
Consumption During Rebalance	STOPPED	CONTINUES on non-revoked partitions
Complexity	Simple (single step)	Complex (multiple steps)
Debugging	Easier to trace	Multi-phase, harder to debug
Consumer Lag Impact	Higher (all partitions pause)	Lower (only moved partitions pause)
State Management	All state reset	Partial state retention
Best For	Small groups, stateless consumers	Large groups, stateful consumers
Good Fit	Infrequent changes, simple systems	Frequent scaling, high-throughput

Static Group Membership

Cooperative rebalancing is great, but what if you don't want any rebalance during brief restarts?

The Problem

By default:

Consumer leaves → Loses member ID
Consumer rejoins → Gets new member ID
Rebalance triggered (even for brief restart)

The Solution: Static Members

Configure consumers with fixed IDs:

properties.setProperty("group.instance.id", "consumer-1");

Behavior

Consumer rejoins within session.timeout.ms:

✅ Keeps same partition assignment
✅ NO rebalance triggered

Consumer away longer than session.timeout.ms:

❌ Rebalance triggered
❌ Partitions reassigned

Use Cases

1. Kubernetes/Container Environments

Pod restarts don't trigger rebalance
Rolling updates happen smoothly

2. Local Cache/State Maintenance

Consumers maintain local state for their partitions
Avoid rebuilding cache on restart
Ensure partition affinity

Key Takeaways

✅ Rebalancing strategies are design choices - not one-size-fits-all
✅ Eager rebalancing is simpler but pauses all consumers
✅ Cooperative rebalancing minimizes disruption but adds complexity
✅ Choose based on your use case - group size, scaling frequency, state management
✅ Multiple strategies in config allows backward compatibility and migration
✅ Static group membership prevents rebalance during brief restarts
✅ Logs reveal the process - watch "Revoked partitions" to understand impact

Conclusion

Understanding rebalancing strategies was a turning point in my Kafka learning journey. Rather than one being "better," each strategy solves different problems.

Key insights:

Eager rebalancing works well for simple, small-scale systems where simplicity matters
Cooperative rebalancing shines in large-scale, stateful, high-throughput scenarios
The "best" strategy depends on your specific requirements and constraints
Static group membership complements both strategies for handling restarts
Real logs help you understand what's actually happening during rebalances

There's no universal "right" answer - choose the strategy that fits your system's characteristics and operational needs.

This article is part of my learning journey through Apache Kafka. If you found it helpful, please give it a like and follow for more Kafka tutorials!

Course Reference: Apache Kafka Series - Learn Apache Kafka for Beginners v3

Learning Apache Kafka with Python - Part 1: Producers

Byron Hsieh — Fri, 02 Jan 2026 09:25:22 +0000

Introduction

After setting up a local Kafka environment with Docker (covered in my previous post), I started converting Java Kafka producer examples to Python as part of my learning process. The Udemy course uses Java, but I want to practice Kafka concepts using Python.

This journal documents the practical challenges, solutions, and patterns I discovered while converting two producer implementations: a basic producer and a producer with key-based partitioning.

Note: For Kafka producer concepts (sticky partitioning, acknowledgments, etc.), see my separate article "Understanding Kafka Producer From Basics to Sticky Partitioning". This journal focuses on the Python implementation journey.

Challenge #1: Translating Java Error Handling to Python

Java Try-Catch-Finally Pattern

Python Translation

Converting this pattern to Python, I learned several important details:

def main():
    producer = None  # ← Must initialize outside try!

    try:
        config = create_producer_config()
        producer = Producer(config)
        producer.produce(topic='demo_python', value='hello'.encode('utf-8'))
        producer.flush()

    except BufferError as e:
        log.error(f"Producer queue full: {e}", exc_info=True)
        raise  # ← Bare raise preserves traceback

    except Exception as e:
        log.error(f"Error: {e}", exc_info=True)
        raise

    finally:
        if producer is not None:
            try:
                producer.flush(timeout=5)
                log.info("Producer closed successfully")
            except Exception as e:
                log.warning(f"Cleanup error: {e}")  # ← Don't crash on cleanup

Key Differences from Java

Aspect	Java	Python
Syntax	try-catch-finally	try-except-finally
Variable scope	Inside try block	Must initialize before try
Specific exceptions	Multiple catch blocks	Multiple except blocks
Traceback logging	`log.error("msg", e)`	`log.error("msg", exc_info=True)`
Re-throw	`throw e;`	`raise` (bare, no argument)

Python-Specific Gotchas I Learned

1. Why Initialize `producer = None` Outside Try?

# ❌ WRONG: Variable undefined if Producer() fails
try:
    producer = Producer(config)  # May fail here
    # ...
finally:
    producer.flush()  # NameError if Producer() failed!

# ✅ CORRECT: Always defined
producer = None
try:
    producer = Producer(config)
    # ...
finally:
    if producer is not None:  # Safe check
        producer.flush()

In Java, variables declared in try block are still accessible in finally. In Python, if the line that defines the variable throws an exception, the variable doesn't exist.

2. Bare `raise` vs `raise e`

except BufferError as e:
    log.error(f"Queue full: {e}", exc_info=True)
    raise  # ← Nothing after raise!

Three ways to raise:

raise - Re-raises exact exception with full traceback ✅
raise e - Re-raises but loses original traceback ❌
raise CustomError() - Raises different exception ⚠️

Why bare raise matters: It shows where the error originally occurred, not just where you re-raised it.

3. Nested Try-Except in Finally

finally:
    if producer is not None:
        try:  # ← Nested try for cleanup
            producer.flush(timeout=5)
        except Exception as e:
            log.warning(f"Cleanup error: {e}")  # Don't re-raise

Cleanup can fail too! Use log.warning() instead of log.error() since we're already shutting down.

Challenge #2: The Mysterious `poll()` Method

The Confusing Part

This was the most confusing difference between Java and Python Kafka clients:

Java (kafka-clients library):

producer.send(record, callback);  // Callback executes automatically

Python (confluent-kafka library):

producer.produce(topic='demo', value='msg', callback=callback)
# Callback WON'T execute yet!

producer.poll(0)  # ← Must explicitly call this
# NOW callback executes

What I Learned

The confluent-kafka library (Python) wraps librdkafka (C library). Unlike Java's client, you must explicitly call poll() to:

Trigger network operations (send queued messages)
Receive acknowledgments from broker
Execute callbacks

producer.produce(...)  # Step 1: Queue in memory (fast)
producer.poll(0)       # Step 2: Send + trigger callback

poll() Variations

producer.poll(0)      # Non-blocking: process ready events, return immediately
producer.poll(1.0)    # Blocking: wait up to 1 second for events
producer.flush()      # Blocking: wait for ALL messages + callbacks

Visual Comparison

Without poll() ❌:

for i in range(10):
    producer.produce(...)  # Queue all 10 messages
    # No poll() - callbacks delayed

producer.flush()  # All 10 callbacks fire here at once

With poll(0) ✅:

for i in range(10):
    producer.produce(...)  # Queue message
    producer.poll(0)       # Callback fires immediately

This gives real-time feedback about partition assignment, which is crucial for the key-partitioning demo!

Challenge #3: Callback Timing - Sync vs Async

The Unexpected Output

When I first ran producer_demo_keys.py, the output was confusing:

--- Round 1 ---

--- Round 2 ---
Key: id_0 | Partition: 1  ← Where's Round 1?
Key: id_3 | Partition: 1
...

All Round 1 callbacks fired during Round 2! This revealed an important decision point.

The Cause

for round_num in range(2):
    for i in range(10):
        producer.produce(...)
        producer.poll(0)  # Non-blocking

    time.sleep(0.5)  # Not long enough for callbacks to complete

Network latency (even on localhost) meant callbacks from Round 1 hadn't executed before Round 2 started.

Two Valid Approaches

I realized both patterns have valid use cases, so I made it configurable:

Approach 1: Synchronized (Educational)

for round_num in range(2):
    for i in range(10):
        producer.produce(...)
        producer.poll(0)

    producer.flush()  # ← Block until all callbacks complete

✅ Clean output showing Round 1 complete before Round 2

✅ Perfect for demos and testing

❌ Lower throughput (blocking waits)

Approach 2: Async (Production)

for round_num in range(2):
    for i in range(10):
        producer.produce(...)
        producer.poll(0)

    time.sleep(0.5)  # ← Non-blocking

✅ Higher throughput

✅ More realistic production behavior

⚠️ Callbacks may interleave

Making It Toggleable

Added configuration comments in the code:

# ============================================================
# TWO APPROACHES: Choose based on your use case
# ============================================================

# APPROACH 1: Synchronized Rounds (Educational/Testing)
producer.flush()  # ← Active by default

# APPROACH 2: Async High-Throughput (Production)
# Comment out flush() above, uncomment below:
# time.sleep(0.5)
# ============================================================

Key insight: Either way, the key-to-partition mapping remains consistent - that's what matters!

Java to Python Conversion Patterns

Configuration: Properties → dict

Java:

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());

Python:

config = {
    'bootstrap.servers': 'localhost:9092',
    # No serializer config needed for simple strings
    # confluent-kafka handles it automatically
}

Producing Messages

Java:

ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic", "key", "value");
producer.send(record);

Python:

producer.produce(
    topic='topic',
    key='key'.encode('utf-8'),    # Must encode to bytes
    value='value'.encode('utf-8')  # Must encode to bytes
)

Key difference: Python's confluent-kafka requires bytes, so always .encode('utf-8').

Callbacks

Java:

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception e) {
        if (e == null) {
            log.info("Partition: " + metadata.partition());
        } else {
            log.error("Error: " + e.getMessage());
        }
    }
});

Python:

def delivery_callback(err, msg):
    if err is not None:
        log.error(f"Delivery failed: {err}")
    else:
        log.info(f"Partition: {msg.partition()}")

producer.produce(topic='topic', value='msg', callback=delivery_callback)
producer.poll(0)  # ← Must explicitly trigger!

Resource Cleanup

Java:

producer.flush();
producer.close();

Python:

producer.flush()
# No explicit close() - flush() is sufficient
# Python handles cleanup via garbage collection

Final Implementations

producer_demo_keys.py (With Partitioning)

import time
from confluent_kafka import Producer, KafkaException
from kafka_logging import get_logger

log = get_logger(__name__)

def delivery_callback(err, msg):
    if err is not None:
        log.error(f"Message delivery failed: {err}")
    else:
        key_str = msg.key().decode('utf-8') if msg.key() else None
        log.info(f"Key: {key_str} | Partition: {msg.partition()}")

def main():
    producer = None

    try:
        config = {'bootstrap.servers': 'localhost:9092'}
        producer = Producer(config)
        topic = 'demo_python'

        for round_num in range(2):
            log.info(f"\n--- Round {round_num + 1} ---")

            for i in range(10):
                key = f"id_{i}"
                value = f"hello world {i}"

                producer.produce(
                    topic=topic,
                    key=key.encode('utf-8'),
                    value=value.encode('utf-8'),
                    callback=delivery_callback
                )
                producer.poll(0)  # Trigger callbacks

            # Choose approach:
            producer.flush()   # Synchronized
            # time.sleep(0.5)  # Async

        log.info("All messages sent!")

    except BufferError as e:
        log.error(f"Queue full: {e}", exc_info=True)
        raise
    except KafkaException as e:
        log.error(f"Kafka error: {e}", exc_info=True)
        raise
    except Exception as e:
        log.error(f"Error: {e}", exc_info=True)
        raise
    finally:
        if producer is not None:
            try:
                producer.flush(timeout=5)
                log.info("Producer closed")
            except Exception as e:
                log.warning(f"Cleanup error: {e}")

if __name__ == "__main__":
    main()

Testing

Running the Producers

# Basic producer
uv run python -m kafka_basics.producer_demo

# Producer with keys
uv run python -m kafka_basics.producer_demo_keys

Verifying with Console Consumer

docker exec -it broker kafka-console-consumer.sh \
  --topic demo_python \
  --from-beginning \
  --bootstrap-server localhost:9092

Press Ctrl+C to exit.

Key Takeaways

Python-Specific Lessons

Bytes Encoding: Always .encode('utf-8') for keys and values
poll() is Required: Unlike Java, callbacks need explicit triggering
Variable Scope: Initialize resources before try block in Python
Bare raise: Use raise without arguments to preserve full traceback

Error Handling Patterns

Initialize resources to None before try block
Use bare raise to preserve tracebacks
Nest try-except in finally block for safe cleanup
Use log.warning() for cleanup errors (not log.error())
Always add timeouts to prevent indefinite blocking

Conversion Strategy

Java Concept	Python Equivalent	Notes
Properties	dict	Use dot-notation keys
try-catch-finally	try-except-finally	Different syntax, same pattern
Anonymous Callback	Function callback	Define as regular function
producer.send()	producer.produce() + poll()	Explicit callback trigger needed
producer.close()	producer.flush()	flush() is sufficient

What I Learned About Python Development

Resource Management: Python's finally block works like Java but with scope differences
Library Quirks: confluent-kafka requires manual poll() unlike Java client
Configuration Flexibility: Python's dict makes config more readable than Java Properties
Async Patterns: Understanding when to use flush() vs poll() for different use cases

Resources

Session Date: January 2, 2026
Python 3.13, uv package manager, confluent-kafka, Kafka 4.0.1

Graceful Shutdown in Kafka: Understanding Shutdown Hooks and Thread Management

Byron Hsieh — Tue, 30 Dec 2025 15:38:11 +0000

Introduction

When I first saw the ConsumerDemoWithShutdown.java code, I was puzzled by this comment:

// get a reference to the main thread
final Thread mainThread = Thread.currentThread();

Why do we need a reference to the main thread? What's a Shutdown Hook? And what does join() actually do?

As a Java and Kafka beginner, these concepts were confusing. But after diving deep into the code, I realized this is one of the most important patterns for building reliable Kafka applications.

In this article, I'll explain everything from Shutdown Hooks to Singleton patterns to Thread.join() - all the foundational concepts you need to understand graceful shutdown.

This guide is based on the excellent course "Apache Kafka Series - Learn Apache Kafka for Beginners v3".

Part 1: The Problem - Why We Need Graceful Shutdown

The Naive Approach (Without Shutdown Hook)

Let's start with a simple consumer:

public class BadConsumer {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("demo_java"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            // Process messages...
        }

        // This line is NEVER reached!
        consumer.close();
    }
}

What happens when you press Ctrl+C?

❌ The while(true) loop is interrupted
❌ consumer.close() is NEVER called
❌ Offsets are NOT committed
❌ Resources are NOT released
❌ Next time you start: duplicate message processing

This is a serious problem in production systems!

Part 2: Understanding Shutdown Hooks

What is a Shutdown Hook?

A Shutdown Hook is a special mechanism provided by the JVM (Java Virtual Machine) that allows you to run cleanup code when your program is about to exit.

Think of it as: "Hey JVM, before you shut down, please run this cleanup code for me!"

When Does a Shutdown Hook Trigger?

✅ It WILL trigger when:

You press Ctrl+C (SIGINT signal)
You call System.exit()
Your program finishes normally
Operating system sends SIGTERM

❌ It will NOT trigger when:

Force kill: kill -9 (SIGKILL)
JVM crashes
Operating system crashes

Basic Shutdown Hook Example

public class ShutdownHookDemo {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("Application starting...");

        // Register a Shutdown Hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Shutdown detected! Cleaning up...");
        }));

        // Simulate work
        System.out.println("Working...");
        Thread.sleep(10000); // Sleep for 10 seconds

        System.out.println("Work done!");
    }
}

Try it yourself:

Run the program
Press Ctrl+C during the 10-second sleep
You'll see: "Shutdown detected! Cleaning up..."

Part 3: Understanding Runtime.getRuntime()

Before diving into the Kafka code, I needed to understand what Runtime.getRuntime() means.

What is Runtime?

The Runtime class represents the JVM (Java Virtual Machine) environment. Since there's only ONE JVM per application, Runtime uses the Singleton pattern - ensuring only one instance exists.

Key point: You can't create a Runtime object with new Runtime(). Instead, you must use:

Runtime runtime = Runtime.getRuntime();

This always returns the same Runtime instance throughout your application.

Why This Matters for Shutdown Hooks

// Register a shutdown hook to the JVM
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    System.out.println("Cleanup code runs here");
}));

Runtime is how we access JVM-level operations like adding shutdown hooks.

Part 4: Understanding Thread.join()

The join() method was initially confusing to me. Here's what I learned.

What Does join() Do?

thread.join() makes the current thread wait until another thread finishes executing.

Simple analogy: It's like waiting for someone to finish their task before you continue.

Basic Example

Thread worker = new Thread(() -> {
    System.out.println("Working...");
    Thread.sleep(2000);
    System.out.println("Done!");
});

worker.start();
worker.join(); // Wait here until worker finishes
System.out.println("Worker completed, continuing...");

Visual Comparison

Without join():

Main Thread:    [start worker] → [END]
Worker Thread:                [Working...] → [END]
                               ↑ Main doesn't wait!

With join():

Main Thread:    [start worker] → [join - waiting...] → [END]
Worker Thread:                [Working...] → [Done!] ┘

Why It Throws InterruptedException

The join() method can be interrupted by other threads, so we need to handle the exception:

try {
    thread.join();
} catch (InterruptedException e) {
    e.printStackTrace();
}

Part 5: The Complete Kafka Shutdown Pattern

Now let's put all the pieces together and understand the full Kafka consumer shutdown code.

Step 1: Get a Reference to the Main Thread

// get a reference to the main thread
final Thread mainThread = Thread.currentThread();

Why do we need this?

The Shutdown Hook runs in a different thread
That thread needs to know which thread to wait for
Thread.currentThread() returns the currently executing thread (in this case, main thread)
final keyword allows the anonymous inner class to access this variable

Step 2: Register the Shutdown Hook

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        log.info("Detected a shutdown, let's exit by calling consumer.wakeup()...");
        consumer.wakeup();

        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

Breaking it down:

Runtime.getRuntime() - Get the singleton Runtime instance
addShutdownHook(new Thread() {...}) - Register a new thread to run on shutdown
new Thread() { public void run() {...} } - Anonymous inner class defining thread behavior
consumer.wakeup() - Wake up the consumer from poll()
mainThread.join() - Wait for main thread to finish cleanup

Step 3: Understanding consumer.wakeup()

// Main thread is stuck here, waiting for messages
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

The poll() method blocks (waits) for up to 1000ms looking for new messages.

What does wakeup() do?

Interrupts the poll() operation
Makes poll() throw WakeupException
Allows the while loop to be exited

Without wakeup():
Main thread might be stuck in poll() for up to 1 second before noticing the shutdown!

Step 4: The Main Consumer Loop

try {
    consumer.subscribe(Arrays.asList(topic));

    while (true) {
        log.info("Polling");
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

        for (ConsumerRecord<String, String> record : records) {
            log.info("key: " + record.key() + ", Value: " + record.value());
            log.info("Partition: " + record.partition() + ", Offset: " + record.offset());
        }
    }
} catch (WakeupException e) {
    log.info("Consumer is starting to shut down");
} catch (Exception e){
    log.error("Unexpected exception in consumer", e);
} finally {
    consumer.close(); // This is GUARANTEED to run
    log.info("The consumer is now gracefully shut down");
}

The flow:

Normal operation: Continuously poll for messages
Ctrl+C pressed: Shutdown Hook calls wakeup()
WakeupException thrown: Caught by the catch block
Finally block: Always executes, closing the consumer

Part 6: Complete Execution Flow

Let's trace what happens when you press Ctrl+C:

Timeline Visualization

┌─ Time ─────────────────────────────────────────────────┐

1. Normal Operation
   Main Thread: [polling...polling...polling...]

2. User Presses Ctrl+C
   ↓
   JVM Detects Shutdown Signal

3. JVM Starts Shutdown Hook Thread
   ┌─ Shutdown Hook Thread ─────────────────┐
   │ 1. log("Detected a shutdown...")       │
   │ 2. consumer.wakeup()  ─────────┐       │
   │ 3. mainThread.join()           │       │
   │    [WAITING...]                │       │
   └────────────────────────────────┼───────┘
                                    │
                                    │ wakeup signal
                                    ↓
   ┌─ Main Thread ─────────────────────────┐
   │ poll() receives wakeup signal         │
   │ → Throws WakeupException              │
   │ → Enters catch block                  │
   │ → log("Consumer is shutting down")    │
   │ → Enters finally block                │
   │ → consumer.close()                    │
   │    - Commits offsets                  │
   │    - Releases resources               │
   │ → log("Gracefully shut down")         │
   │ → Main thread ENDS ──────────┐        │
   └──────────────────────────────┼────────┘
                                  │
                                  │ main thread finished
                                  ↓
   ┌─ Shutdown Hook Thread ────────────────┐
   │ join() returns                        │
   │ Shutdown Hook thread ENDS             │
   └───────────────────────────────────────┘

4. JVM Exits Cleanly
   All resources properly released! ✓

└────────────────────────────────────────────────────────┘

Why join() is Critical

Without join():

Shutdown Hook:  [wakeup()] → [END]
                     ↓
Main Thread:        [processing...] → [close()] → [END]
                                      ↑
                                   Might not finish!
                                   JVM exits too early!

With join():

Shutdown Hook:  [wakeup()] → [join() - WAITING...] → [END]
                     ↓                        ↑
Main Thread:        [processing] → [close()] ┘ → [END]
                                  ↑
                              Guaranteed to complete!

Part 7: My Understanding - How It All Connects

After going through all the concepts above, here's how I finally understood the complete shutdown mechanism:

The Shutdown Hook is a JVM Mechanism

I learned that the shutdown hook is provided by the JVM itself. When I press Ctrl+C (or when the program exits in other ways), the JVM triggers this hook. The run() function inside the shutdown hook is what gets executed when this trigger happens.

What Happens in the run() Method

In the shutdown hook's run() method, two key things happen:

consumer.wakeup() is called - This interrupts the consumer that's stuck in the infinite polling loop
mainThread.join() is called - This makes the shutdown hook thread wait for the main thread to finish

The Try-Catch-Finally Structure

The infinite polling loop is wrapped with a try-catch-finally structure:

try {
    while (true) {
        consumer.poll(...);  // Infinite loop polling for messages
    }
} catch (WakeupException e) {
    // Catches the exception thrown by consumer.wakeup()
    log.info("Consumer is starting to shut down");
} finally {
    consumer.close();  // This ALWAYS executes
}

How the Exception Flow Works

Here's the key insight that helped me understand the flow:

The shutdown hook's run() calls consumer.wakeup()
This causes consumer.poll() to throw a WakeupException
The exception breaks out of the infinite while(true) loop
The catch block catches WakeupException and logs "...starting to shut down..."
The finally block ALWAYS executes and calls consumer.close()

Why This Design Makes Sense

The way I see it now:

Without the shutdown hook: The infinite loop would never break, consumer.close() would never run
Without wakeup(): The main thread would be stuck in poll(), not knowing it needs to shut down
Without join(): The JVM might exit before consumer.close() finishes, losing uncommitted offsets
Without try-catch-finally: We couldn't handle the WakeupException properly and guarantee cleanup

This pattern ensures that no matter how the program exits (Ctrl+C, System.exit(), etc.), the consumer will always close gracefully.

Part 8: Key Concepts Summary

Shutdown Hook

Aspect	Details
Purpose	Execute cleanup code before JVM exits
Registration	`Runtime.getRuntime().addShutdownHook(thread)`
Triggers	Ctrl+C, System.exit(), normal termination
Does NOT trigger	kill -9, JVM crash

Singleton Pattern

Aspect	Details
Definition	Ensures only ONE instance of a class exists
Implementation	Private constructor + static getInstance()
Example	Runtime class
Why	Some resources should be unique (JVM environment)

Thread.join()

Aspect	Details
Purpose	Wait for another thread to complete
Syntax	`thread.join()` or `thread.join(timeout)`
Throws	InterruptedException (if interrupted while waiting)
Use in Kafka	Ensure main thread completes cleanup before JVM exits

consumer.wakeup()

Aspect	Details
Purpose	Interrupt a consumer that's blocked in poll()
Effect	Throws WakeupException in the polling thread
Thread-safe	Can be called from a different thread
Use case	Graceful shutdown from Shutdown Hook

Part 9: Common Mistakes and Troubleshooting

Mistake 1: Not Using final for mainThread

// ❌ Wrong - Compiler error!
Thread mainThread = Thread.currentThread();

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        mainThread.join(); // Error: Cannot access non-final variable
    }
});

// ✅ Correct
final Thread mainThread = Thread.currentThread();

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        mainThread.join(); // Works!
    }
});

Why? Anonymous inner classes can only access final or effectively final variables from the enclosing scope.

Mistake 2: Calling wakeup() From Main Thread

// ❌ Wrong - This doesn't help!
while (true) {
    consumer.wakeup(); // This is in the SAME thread!
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
}

wakeup() must be called from a different thread (like a Shutdown Hook) to interrupt poll().

Mistake 3: Forgetting the catch Block

// ❌ Wrong - WakeupException propagates!
try {
    while (true) {
        consumer.poll(Duration.ofMillis(1000));
    }
} finally {
    consumer.close();
}

// ✅ Correct - Catch WakeupException
try {
    while (true) {
        consumer.poll(Duration.ofMillis(1000));
    }
} catch (WakeupException e) {
    // Expected exception - handle gracefully
} finally {
    consumer.close();
}

Mistake 4: Not Handling InterruptedException

// ❌ Wrong - Ignoring the exception
mainThread.join();

// ✅ Correct - Always handle it
try {
    mainThread.join();
} catch (InterruptedException e) {
    e.printStackTrace();
}

Key Takeaways

✅ Shutdown Hooks provide a way to run cleanup code before JVM exits
✅ Runtime is a Singleton - there's only one JVM environment per application
✅ Thread.join() makes one thread wait for another to complete
✅ consumer.wakeup() interrupts poll() from a different thread
✅ final keyword is necessary for variables accessed in anonymous inner classes
✅ try-catch-finally pattern ensures resources are always released
✅ Graceful shutdown prevents data loss and duplicate processing

Conclusion

When I started learning Kafka, I didn't understand why we needed all this complexity just to stop a consumer. But now I realize that graceful shutdown is fundamental to building reliable systems.

The key insights:

Shutdown Hooks give you a chance to cleanup before the JVM exits
Singleton pattern (like Runtime) ensures system resources are managed correctly
Thread coordination (join, wakeup) allows different threads to work together
Proper exception handling ensures cleanup code always runs

This pattern isn't just for Kafka - it applies to any Java application that needs to cleanup resources on shutdown: database connections, file handles, network sockets, and more.

Understanding these fundamentals will make you a better Java developer and help you build more robust applications.

This article is part of my learning journey through Apache Kafka. If you found it helpful, please give it a like and follow for more Kafka tutorials!

Course Reference: Apache Kafka Series - Learn Apache Kafka for Beginners v3

Setting Up Kafka 4.0 Locally with Docker: A Learning Journey

Byron Hsieh — Tue, 23 Dec 2025 08:23:05 +0000

Introduction

As part of my journey through the "Apache Kafka Series - Learn Apache Kafka for Beginners v3" Udemy course, I needed to set up a local Kafka environment using Docker. What started as a simple container setup evolved into a deeper understanding of Kafka architecture, Docker best practices, and the transition from Zookeeper to KRaft mode.

In this post, I'll share my learning process, the decisions I made, and the final production-ready configuration I arrived at.

Starting Point: Finding the Right Docker Configuration

Initial Research - Confluent vs Bitnami vs Apache Official

When searching for Kafka Docker setups, I encountered three main options:

Confluent's official tutorial - https://developer.confluent.io/confluent-tutorials/kafka-on-docker/
Bitnami Kafka image - Popular for its ease of use
Apache Kafka official image - The source of truth

Initially, I was torn between Bitnami (known for simplified configuration) and Apache's official image (more control but steeper learning curve).

The Confluent Tutorial Discovery

The Confluent tutorial provided an excellent starting point with this configuration:

services:
  broker:
    image: apache/kafka:latest
    hostname: broker
    container_name: broker
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

Why I chose the Apache official image:

✅ Direct from the source (Apache Foundation)
✅ Production-ready and enterprise-grade
✅ Better alignment with official documentation
✅ Latest features and security updates

Key Learning #1: Understanding KRaft Mode

One of the biggest revelations was learning about KRaft (Kafka Raft) - Kafka's replacement for Zookeeper.

What is KRaft?

Kafka Raft Algorithm for Tracking metadata
Eliminates Zookeeper dependency
Single process can handle both broker and controller roles
Faster startup and simpler architecture

Configuration Breakdown:

KAFKA_PROCESS_ROLES: broker,controller        # Single node handles both roles
KAFKA_NODE_ID: 1                             # Unique node identifier
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093  # Controller election
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER   # Controller communication
CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk            # Unique cluster identifier

Benefits of KRaft mode:

⚡ Faster startup (no Zookeeper coordination)
🏗️ Simpler architecture
📈 Better scalability
🔄 Single point of configuration

Key Learning #2: Docker Image Layering and Customization

The Command Path Problem

Initially, running Kafka commands required full paths:

docker exec broker /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

This was verbose and error-prone. I learned about Docker image layering and decided to create a custom image.

Solution: Custom Dockerfile

FROM apache/kafka:4.0.1

# Add Kafka bin directory to PATH for convenient command usage
ENV PATH="/opt/kafka/bin:${PATH}"

# Set working directory
WORKDIR /opt/kafka

Updated docker-compose.yml

services:
  broker:
    build:
      context: .
      dockerfile: Dockerfile
    image: my-kafka-kraft:4.0.1               # Custom image name
    # ... rest of configuration

Key insights:

✅ Original Apache image remains unchanged
✅ New layer adds convenience without bloat
✅ Commands now work directly: kafka-topics.sh --list --bootstrap-server localhost:9092

Key Learning #3: Understanding Kafka Listeners

The listener configuration was initially confusing but crucial for proper networking:

KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092

LISTENERS vs ADVERTISED_LISTENERS:

LISTENERS = "Where Kafka actually listens" (Server-side binding)

ADVERTISED_LISTENERS = "How clients should connect" (Client-side addressing)

Listener Breakdown:

Listener	Binding Address	Port	Purpose	Access From
`PLAINTEXT://broker:29092`	Container hostname	29092	Inter-service communication	Docker network
`CONTROLLER://broker:29093`	Container hostname	29093	KRaft metadata operations	Internal only
`PLAINTEXT_HOST://0.0.0.0:9092`	All interfaces	9092	External client access	Host machine

Why Different Addresses?

Internal Communication: broker:29092
- Other Docker services connect using container hostname
- Fast, low-latency container-to-container networking
Controller Operations: broker:29093
- KRaft protocol for cluster coordination
- Replaces Zookeeper functionality
External Access: localhost:9092
- Host machine applications connect via port forwarding
- Docker maps container port to host port

Network Flow Diagram:

External App → localhost:9092 → Docker Port Mapping → PLAINTEXT_HOST://0.0.0.0:9092
                                                           ↓
Internal Service → broker:29092 → PLAINTEXT://broker:29092 → Kafka Broker
                                                           ↓
KRaft System → broker:29093 → CONTROLLER://broker:29093 ↗

Key Insight: The address Kafka binds to (0.0.0.0) differs from what it advertises to clients (localhost) because clients can't connect to 0.0.0.0 directly.

Key Learning #4: Data Persistence Strategy

The Problem: Data Loss on Container Restart

Initially, running docker-compose down would delete all topics and data. This happened because:

environment:
  KAFKA_LOG_DIRS: /tmp/kraft-combined-logs  # Data stored in container's temp directory
# No volumes configured = data loss on container deletion

Solution: Docker Volumes

services:
  broker:
    # ... other configuration
    volumes:
      - ./data:/tmp/kraft-combined-logs  # Bind mount for data persistence

What Gets Persisted:

./data/
├── __cluster_metadata-0/              # KRaft metadata (replaces Zookeeper)
├── __consumer_offsets-*/              # Consumer group offsets
├── my-topic-0/                        # Topic partition data
│   ├── 00000000000000000000.log       # Actual messages
│   ├── 00000000000000000000.index     # Message index
│   └── partition.metadata             # Partition metadata
├── meta.properties                    # Broker metadata
└── ...                               # Other Kafka state files

Volume strategy comparison:

Method	Pros	Cons	Use Case
No volumes	Simple setup	Data loss on restart	Learning/testing only
Named volumes	Docker managed	Hidden location	Development
Bind mounts	Full control, easy backup	Manual directory management	Production

Final Local Development Configuration

Here's my complete local development setup:

docker-compose.yml

services:
  broker:
    build:
      context: .                              # Build context
      dockerfile: Dockerfile                  # Custom Dockerfile
    image: my-kafka-kraft:4.0.1              # Explicit image name
    hostname: broker
    container_name: broker
    ports:
      - 9092:9092                            # External client port
    volumes:
      - ./data:/tmp/kraft-combined-logs      # Data persistence
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller  # KRaft mode
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

Dockerfile

FROM apache/kafka:4.0.1

# Add Kafka bin directory to PATH for convenient command usage
ENV PATH="/opt/kafka/bin:${PATH}"

# Set working directory
WORKDIR /opt/kafka

Project Structure

kafka-docker/
├── docker-compose.yml      # Main configuration
├── Dockerfile              # Custom image definition
├── data/                   # Kafka data (auto-created)
│   ├── __cluster_metadata-0/
│   ├── my-topic-0/
│   └── ...
└── docker_commands         # Command reference file

Key Takeaways and Best Practices

1. Choose the Right Base Image

Use official Apache Kafka for production environments
Bitnami is excellent for quick prototyping
Always pin specific versions: apache/kafka:4.0.1 vs latest

2. Embrace KRaft Mode

Simpler than Zookeeper-based setups
Better performance and reliability
Future-proof (Zookeeper deprecation planned)

3. Layer Docker Images Thoughtfully

Keep customizations minimal and purpose-driven
Document why each layer exists
Use multi-stage builds for complex setups

4. Plan for Data Persistence

Always use volumes in production
Bind mounts offer better control than named volumes
Backup strategy should include volume data

5. Network Configuration Matters

Understand internal vs external listeners
Plan port allocation carefully
Test connectivity from both inside and outside containers

Conclusion

This journey from a simple Docker container to a well-configured local Kafka setup taught me valuable lessons about:

Modern Kafka architecture (KRaft vs Zookeeper)
Docker best practices (layering, volumes, networking)
Configuration decisions (persistence, networking, image customization)
Development environment setup (network restrictions, data management)

The final configuration is suitable for local development and learning, with a solid foundation that could be enhanced for production use when needed.

Resources

Understanding Kafka Producer: From Basics to Sticky Partitioner

Byron Hsieh — Sun, 21 Dec 2025 10:18:02 +0000

Introduction

When I started learning Apache Kafka, one of the first questions I had was: "How exactly does a Producer work?" The documentation was thorough, but I wanted to understand it through practical examples.

In this article, I'll walk you through what I learned about Kafka Producers, from sending your first message to understanding the mysterious "Sticky Partitioner." We'll build two simple Java programs and explore some interesting behaviors along the way.

This guide is based on the excellent course "Apache Kafka Series - Learn Apache Kafka for Beginners v3".

Part 1: My First Kafka Producer

Let's start with the simplest possible Kafka Producer that sends a single message.

Setting Up Producer Properties

Every Kafka Producer needs configuration. We use Java's Properties class for this:

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());

What's happening here?

bootstrap.servers: The address of your Kafka broker(s)
key.serializer: Converts your key object into bytes for transmission
value.serializer: Converts your value object into bytes

Note: java.util.Properties is similar to Python's dict, but it only stores string key-value pairs. It's part of the Java standard library, not specific to Kafka.

Creating the Producer

With properties configured, we can create our producer:

KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

The generic types <String, String> represent the types for key and value respectively.

Creating and Sending a Message

ProducerRecord<String, String> producerRecord =
    new ProducerRecord<>("demo_java", "hello world");

producer.send(producerRecord);

Here, "demo_java" is the topic name, and "hello world" is our message. Notice we didn't specify a key - it defaults to null.

The Critical Trio: send(), flush(), close()

This is where things get interesting. Here's what each method does:

send() - Asynchronous Operation

producer.send(producerRecord);

Asynchronous: The message goes into a buffer, it's NOT sent immediately
If your program exits right after this, the message might never reach Kafka

flush() - Synchronous Operation

producer.flush();

Synchronous: Forces all buffered messages to be sent and blocks until complete
Useful for learning/demos to ensure messages are sent
Rarely used in production because it impacts performance

close() - Cleanup

producer.close();

Shuts down the Producer and releases resources
Internally calls flush() to ensure all messages are sent
MUST be called in production to prevent resource leaks

Here's the relationship:

send()          flush()         close()
  ↓               ↓               ↓
[Buffer] -----> [Send] -----> [Clean up]
(async)        (sync)         (includes flush)

Best Practice:

Always call close() in production
Avoid calling flush() unless absolutely necessary
Let Kafka handle batching automatically for better performance

Part 2: Producer with Callbacks

Now let's level up and add callbacks to track message metadata.

Why Use Callbacks?

Callbacks let you know when a message is successfully sent (or if it failed):

producer.send(producerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception e) {
        if (e == null) {
            // Success!
            log.info("Received new metadata \n" +
                    "Topic:\t" + metadata.topic() + "\n" +
                    "Partition:\t" + metadata.partition() + "\n" +
                    "Offset:\t" + metadata.offset() + "\n" +
                    "Timestamp:\t" + metadata.timestamp());
        } else {
            // Something went wrong
            log.error("Error while producing", e);
        }
    }
});

What metadata can you get?

topic: Which topic the message was sent to
partition: Which partition number within that topic
offset: The position of this message in the partition
timestamp: When the message was created

Part 3: Understanding the Sticky Partitioner

This is where it gets really interesting. When I ran my producer sending 100 messages to a topic with 3 partitions, I noticed something odd: all messages went to partition 0.

Was my code broken? Not quite. I needed to understand the Sticky Partitioner.

What is the Sticky Partitioner?

Introduced in Kafka 2.4+, the UniformStickyPartitioner is the default partitioner when messages don't have a key.

How it works:

Messages "stick" to one partition until a batch is full
When the batch is sent, switch to a different partition
Goal: Improve performance by reducing network requests

When Does a Batch Get Sent?

A batch is sent when ANY of these conditions is met:

batch.size - Batch reaches configured size (in bytes)
linger.ms - Time limit reached (in milliseconds)
flush() - Manually forced

Common Misconception: batch.size is NOT time!

I initially thought batch.size was related to time. It's not!

batch.size = Size in bytes (default: 16384 bytes or 16 KB)
linger.ms = Time in milliseconds (default: 0)

Demonstrating Sticky Partitioner Behavior

To observe partition switching, we need to make batches fill up quickly:

// Make batches smaller so they fill up faster
properties.setProperty("batch.size", "400");  // 400 bytes instead of 16KB

Why 400?

Each message is roughly 15-20 bytes ("hello worldXX")
400 bytes ÷ 20 bytes ≈ 20 messages per batch
When a batch fills up → it's sent → switches to next partition

The Loop with Delay

for (int i = 0; i < 100; i++) {
    if (i % 10 == 0) {
        Thread.sleep(1000);  // Pause every 10 messages
    }

    ProducerRecord<String, String> producerRecord =
        new ProducerRecord<>("demo_java", "hello world" + i);

    producer.send(producerRecord, callback);
}

Expected Behavior

With 3 partitions and batch.size=400:

First ~20 messages → Partition 0
Next ~20 messages → Partition 1
Next ~20 messages → Partition 2
Cycle continues...

You can observe this in the callback logs showing the partition number!

Key Configuration Parameters

Here's a quick reference table:

Property	Description	Default	Unit
`bootstrap.servers`	Kafka broker address	None	-
`key.serializer`	Key serializer class	None	-
`value.serializer`	Value serializer class	None	-
`batch.size`	Max batch size	16384	bytes
`linger.ms`	Max wait time before sending	0	milliseconds

Keyed vs Keyless Messages

The partitioning behavior changes based on whether you specify a key:

Scenario	Partitioner Behavior
With key	Messages with the same key go to the same partition (hash-based)
Without key	Uses Sticky Partitioner (batch-based)

Troubleshooting Common Issues

Error: Invalid value null for configuration value.serializer

Cause: Typo in property name

❌ values.serializer (with 's')
✅ value.serializer (singular)

All Messages Going to Partition 0

Possible causes:

Topic only has 1 partition
batch.size is too large - all messages fit in one batch
Wrong delay logic (i/10 instead of i%10)

Solution:
Check your topic configuration:

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic demo_java

Adjust batch size:

properties.setProperty("batch.size", "400");

Fix the delay logic:

if (i % 10 == 0) {  // Correct!
    Thread.sleep(1000);
}

Why Sticky Partitioner is Better Than Round-Robin

Old Way: Round-Robin (Pre-Kafka 2.4)

Each message goes to a different partition in sequence
100 messages = potentially 100 network requests
More overhead, more latency

New Way: Sticky Partitioner (Kafka 2.4+)

Messages stick to one partition until batch is full
100 messages = maybe 5 network requests (assuming 20 messages per batch)
Fewer network calls = better throughput

When to Use flush()

Good use cases for flush():

Financial transaction systems (need immediate confirmation)
Critical logging where every message must be guaranteed

Avoid flush() in:

Regular log aggregation
High-throughput data pipelines
Real-time analytics (slight delay is acceptable)

Remember: Kafka's automatic batching is designed for performance. Only override it when you have a specific requirement!

Key Takeaways

✅ Properties configuration is the foundation - get it right
✅ send() is async - use flush() or close() to ensure delivery
✅ Callbacks provide metadata - use them to track message status
✅ Sticky Partitioner batches by partition - improves performance
✅ batch.size is in bytes, not time - linger.ms is the time setting
✅ Keyless messages use Sticky Partitioner
✅ Always call close() - prevents resource leaks

Conclusion

Understanding how Kafka Producers work is fundamental to building robust event-driven systems. The key insights I gained were:

The asynchronous nature of send() and why close() is critical
How callbacks provide visibility into message delivery
The performance benefits of the Sticky Partitioner
The importance of proper configuration (especially batch.size vs linger.ms)

These concepts form the foundation for more advanced Kafka patterns like transactions, idempotent producers, and exactly-once semantics.

This article is part of my learning journey through Apache Kafka. If you found it helpful, please give it a like and follow for more Kafka tutorials!

Course Reference: Apache Kafka Series - Learn Apache Kafka for Beginners v3

From 8 Minutes to 40 Seconds: Solving Data Pipeline Deployment Bottlenecks with Git Sparse Checkout

Byron Hsieh — Thu, 06 Nov 2025 04:40:00 +0000

Introduction

Modern data engineering projects typically use Python for orchestration (Airflow DAGs), data transformation, and DBT or SQL for data ingestion. At scale, however, deployment becomes a significant bottleneck.

My data engineering team manages ingestion for a data warehouse containing nearly 10,000 tables. We follow a standardized approach where each table requires at least 5 programs covering the standard pipeline: file ingestion → staging → transformation → ODS layer.

This results in over 50,000 program files requiring deployment.

The 8-Minute Deployment Problem

Our Azure DevOps CI/CD pipeline was taking nearly 8 minutes per deployment — unacceptable for any development workflow. Having previously managed deployment pipelines for Java microservices with comprehensive test suites, this was excessive for our relatively simple process in my opinion.

Root Cause Analysis

The deployment process consisted of four straightforward steps:

Checkout codebase from repository
Extract programs listed in deployment manifest
Package selected programs
Deploy to target server directories

Investigation revealed the bottleneck: checkout consumed over 7 minutes — over 80% of total deployment time.

With 50,000+ files in our monorepo, Git was downloading the entire codebase even when we only needed a small subset for deployment. This led me to explore Git sparse checkout as a solution.

Understanding Sparse Checkout

Git sparse checkout allows you to download only specific files or directories from a repository, rather than cloning the entire codebase. Introduced in Git 2.25 (2020), it's designed for exactly our use case: large monorepos where you only need a subset of files.

The Traditional Problem

Azure DevOps' default checkout task doesn't support sparse checkout natively. The standard approach looks like this:

steps:
  - checkout: self
    fetchDepth: 1  # Only helps with history, not file count

Even with fetchDepth: 1, Git still downloads all 50,000+ files from our repository. Shallow clones reduce history but don't reduce the working tree size.

Our Deployment Reality

In our data engineering workflow, deployments are selective:

Production deployment of 15 modified DAGs
Staging deployment of 3 new data models
Hotfix deployment of 2 SQL transformations

We don't need all 50,000 files — we need the specific files listed in our deployment manifest.

The Four-Layer Optimization

To maximize performance, we combine sparse checkout with three other Git optimizations:

Blobless clone (--filter=blob:none) — Download tree structure, not file contents initially
Shallow clone (--depth 1) — Skip commit history
Single branch (--single-branch) — Ignore other branches
Sparse checkout (non-cone mode) — Get only files in deployment manifest

Together, these techniques reduce our checkout from 8 minutes to under 40 seconds.

Implementation Guide

The complete implementation is available in my GitHub demo repository. Here's the step-by-step breakdown:

Step 1: Repository Structure

The demo simulates a realistic financial data platform with 293 files:

financial-data-pipeline-demo/
├── dags/                        # 65 Airflow DAGs
├── data_model/                  # 67 DDL files (staging/marts/dimensions)
├── dbt/models/                  # DBT transformation models
├── metadata/                    # 27 schemas & configs
├── deployment/
│   └── deploy-list.txt         # Deployment manifest (10 files)
└── azure-pipeline-sparse-checkout.yml

Step 2: Create the Deployment Manifest

The deployment/deploy-list.txt defines exactly which files to deploy:

dags/stock_price_ingestion.py
dags/bond_yield_analysis.py
dags/forex_rates_pipeline.py
metadata/schemas/stock_price_schema.json
metadata/schemas/bond_yield_schema.json
data_model/staging/stg_stock_prices.sql
data_model/staging/stg_bond_yields.sql
dbt/models/staging/stg_stock_prices.sql
dbt/models/staging/stg_bond_yields.sql
dbt/models/marts/fact_daily_prices.sql

Result: Only 10 files are checked out from 293 total (97.8% reduction).

Step 3: The Sparse Checkout Pipeline

The core optimization is in azure-pipeline-sparse-checkout.yml:

steps:
- checkout: none  # Disable default Azure DevOps checkout

- script: |
    echo "=== Sparse Checkout: Selective File Download ==="

    # Four-layer optimization
    git clone \
      --filter=blob:none \      # Blobless clone
      --no-checkout \            # Don't materialize files yet
      --depth 1 \                # Shallow clone (no history)
      --single-branch \          # Only current branch
      --branch $(Build.SourceBranchName) \
      $(repositoryUrl)

    cd $(Build.SourcesDirectory)

    # Enable file-level sparse checkout (non-cone mode)
    git sparse-checkout init --no-cone

    # Two-stage checkout process
    # Stage 1: Get only the manifest file
    echo "$(deployListFile)" > .git/info/sparse-checkout
    git checkout $(Build.SourceBranchName)

    # Stage 2: Read manifest and checkout actual files
    grep -v '^#' $(deployListFile) | grep -v '^$' > .git/info/sparse-checkout
    git checkout $(Build.SourceBranchName)

    # Performance tracking
    TOTAL_FILES=$(find . -type f -not -path './.git/*' | wc -l)
    echo "Files checked out: $TOTAL_FILES"

  displayName: 'Sparse Checkout - Selective Download'

Step 4: Verification & Validation

The pipeline automatically verifies the sparse checkout worked correctly:

- script: |
    cd $(Build.SourcesDirectory)

    echo "=== Verifying Sparse Checkout ==="
    missing_count=0
    found_count=0

    # Check each file from deploy-list.txt exists
    while IFS= read -r file_path; do
      if [ -f "$file_path" ]; then
        found_count=$((found_count + 1))
      else
        echo "Missing: $file_path"
        missing_count=$((missing_count + 1))
      fi
    done < <(grep -v '^#' $(deployListFile) | grep -v '^$')

    echo "Files found: $found_count"
    echo "Files missing: $missing_count"

  displayName: 'Verify Sparse Checkout'

Key Implementation Details

Why `--no-cone` Mode?

Traditional cone mode only works with directories:

# Cone mode (directory-only)
git sparse-checkout set dags/ metadata/

Non-cone mode enables file-level precision:

# Non-cone mode (file-level)
git sparse-checkout init --no-cone
cat deploy-list.txt > .git/info/sparse-checkout

This lets us cherry-pick specific files across multiple directories.

Why Two-Stage Checkout?

The manifest file itself must be checked out before we can read it:

First checkout: Get deployment/deploy-list.txt
Second checkout: Use manifest contents to get actual files

Without this, the pipeline would fail with "file not found" errors.

Performance Results

Running the pipeline on the demo repository:

Metric	Full Checkout	Sparse Checkout	Improvement
Files Downloaded	293 files	10 files	97.8% reduction
Checkout Time	~45-60s	~5-10s	80-90% faster
Disk Usage	Full repo	Minimal	Significant savings
Network Transfer	All objects	Blob-less + selective	90%+ reduction

Production Environment Impact

In our real-world deployment with 50,000+ program files:

Before Optimization:

Deployment time: 8 minutes per deployment
Checkout phase: 6+ minutes (83% of total time)
Files downloaded: All 50,000+ files every time
Network transfer: ~2GB per deployment

After Sparse Checkout:

Deployment time: 40 seconds (90% reduction)
Checkout phase: <20 seconds (93% improvement)
Files downloaded: ~500 files (only what's needed)
Network transfer: ~200MB (90% reduction)

When to Use This Approach

✅ Ideal for:

Large monorepos (1,000+ files)
Selective deployments (deploying subset of changed files)
Frequent deployments with small change sets
Self-hosted agents with disk constraints
High network transfer costs

❌ Not recommended for:

Small repositories (<100 files)
Full application deployments requiring all files
First-time repository setups
Teams unfamiliar with Git sparse checkout

Conclusion

By combining four optimization techniques—partial clone, shallow fetch, single-branch, and file-level sparse checkout—with sparse checkout being the most critical for selective deployments, we achieved:

90% faster deployments
97.8% fewer files downloaded
90% reduction in network bandwidth
10x improvement in developer productivity

The complete working implementation is available on GitHub for you to try.

Have you faced similar deployment bottlenecks in your data engineering pipelines? Share your experiences in the comments!

DEV Community: Byron Hsieh

Why Data Teams Need Data Lineage: From Common Pain Points to Real-World Challenges

Why Data Teams Need Data Lineage

1. Why Data Teams Need Data Lineage (General)

Complex dependencies make incident investigation expensive

Risky schema and logic changes without impact analysis

Slow onboarding and weak knowledge transfer

Data teams become a support center for business users

Inefficient Data Quality (DQ) incident handling and reprocessing

2. Real-World Challenges: A Highly Heterogeneous SQL Ecosystem

Thousands of legacy batch jobs

SQL is not pure SQL: templates, embedded code, and vendor dialects

Noise from comments, multilingual content, and inconsistent naming

SQL patterns too complex for off-the-shelf lineage tools

A preprocessing pipeline becomes required

3. Conclusion

dbt + OpenLineage #1: Why dbt-ol Is a Post-Processor (Not a Plugin) — and Why It Matters

Introduction

The Setup

Tech Stack

Project Architecture

Key Learning #1: How dbt-ol Actually Works

The Two Artifacts

Running It

Key Learning #2: Anatomy of an OpenLineage Event

The Three Core Entities

Why the parent Facet Matters

Key Learning #3: Column-Level Lineage in Practice

The CTE Alias Problem

What's Missing: The Empty inputs[]

What catalog.json actually changes

What actually changes with catalog.json

Takeaways

What's Next

Resources

Why Multi-Agent Deployment Might Slow Down Your Redshift DDL Deployments

Background

The Deployment Pattern

Azure DevOps Multi-Agent: Understanding the Mechanism

Classic Editor (Release Pipelines)

YAML Pipelines

How to Actually Distribute Work

In Classic Editor (Release Pipelines)

In YAML Pipelines

The Critical Understanding

Redshift Architecture: The Leader Node Constraint

How Redshift Handles DDL

System Catalog Table Locking

Control Table Locking

Performance Analysis: What Would Actually Happen

Scenario 1: Multi-Agent Without Work Distribution

Scenario 2: Multi-Agent With Correct Work Distribution

Scenario 3: Single-Agent Sequential (Current Approach)

Performance Comparison Matrix

Decision: Stay with Single-Agent

Lessons Learned

1. Architecture Understanding Matters More Than Tooling

2. Challenge "Obvious" Solutions

3. Analyze Before Implementing

4. Simple Can Be Better

Key Takeaways

References

Official Documentation

Key Documentation Insights

Kafka Producer Deep Dive: From Basics to Production-Ready Configuration

Introduction

Table of Contents

1. Producer Basics

Setting Up a Producer

The Critical Trio: send(), flush(), close()

send() - Asynchronous Operation

flush() - Synchronous Operation

close() - Cleanup

Callbacks for Monitoring

2. The Sticky Partitioner

How It Works

When Does a Batch Get Sent?

Common Misconception

Why Sticky is Better Than Round-Robin

3. Producer Acknowledgements (acks)