DEV Community: Vivek Burman

How to Data Engineer the ETLFunnel Way

Vivek Burman — Sat, 01 Nov 2025 16:26:00 +0000

Part 4 — Mastering Pipeline Termination

When Your Pipeline Needs to Know When to Stop

Picture this: You've built a beautiful streaming data pipeline. Data flows in smoothly, transformations happen like clockwork, and everything seems perfect. Then you check your cloud bill at the end of the month and realize your pipeline has been running 24/7, processing mostly empty queues during off-peak hours. Sound familiar?

This is where Termination Rules come in — one of ETLFunnel's most powerful features for managing long-running and streaming pipelines.

The Problem: Infinite Pipelines in a Finite World

Traditional batch pipelines have it easy. They process all available data and exit gracefully when done. But modern data engineering often involves:

Streaming pipelines that continuously listen for new events
Pub/Sub consumers waiting for messages that might never come
Event-driven workflows that need to respond to sporadic triggers
Real-time processors that should adapt to business hours

These pipelines don't naturally terminate. They run forever — or until something crashes, resources run out, or someone manually kills them. That's wasteful, expensive, and frankly, inelegant.

Enter Termination Rules

Termination Rules in ETLFunnel provide intelligent, configurable exit conditions for your pipelines. Think of them as the "smart shutdown" mechanism that knows when continuing is pointless or unproductive.

Instead of letting your pipeline run indefinitely, you define clear conditions for when it should gracefully exit:

func TerminateRule(param *models.TerminateRuleProps) (*models.TerminateRuleTune, error) {
    maxRecords := uint64(10000)
    idleTimeout := 5 * time.Minute

    return &models.TerminateRuleTune{
        MaxRecords:    &maxRecords,
        IdleTimeout:   &idleTimeout,
        CheckInterval: 10 * time.Second,
    }, nil
}

This simple rule says: "Stop after processing 10,000 records OR after 5 minutes of no activity — whichever comes first."

The Four Pillars of Termination

ETLFunnel gives you four powerful mechanisms to control pipeline termination:

1. MaxRecords: The Volume Limit

Perfect for: Development, testing, sampling large datasets

maxRecords := uint64(1000)
return &models.TerminateRuleTune{
    MaxRecords: &maxRecords,
    CheckInterval: 5 * time.Second,
}

Use case: You're testing a new transformation on production data. Instead of processing millions of records, you sample the first 1,000 to verify everything works correctly.

2. IdleTimeout: The Silence Detector

Perfect for: Event-driven pipelines, off-peak optimization

idleTimeout := 10 * time.Minute
return &models.TerminateRuleTune{
    IdleTimeout: &idleTimeout,
    CheckInterval: 30 * time.Second,
}

Use case: Your pipeline consumes from a message queue. During nights and weekends, new messages are rare. Why keep the pipeline running? If nothing arrives for 10 minutes, shut it down gracefully.

3. MaxPipelineTime: The Duration Guard

Perfect for: Cost control, scheduled batch windows

maxPipelineTime := 2 * time.Hour
return &models.TerminateRuleTune{
    MaxPipelineTime: &maxPipelineTime,
    CheckInterval: 20 * time.Second,
}

Use case: You have a 2-hour processing window during off-peak hours. Regardless of how much data is available, the pipeline must complete within this window to avoid impacting other systems.

4. UserDefinedCheckFunc: The Custom Logic

Perfect for: Complex business rules, adaptive behavior

This is where termination rules become truly powerful. You can implement any custom logic based on pipeline state:

UserDefinedCheckFunc: func(checkProps *models.CustomTerminateRuleCheckProps) (*models.TerminateRuleActionTune, error) {
    // Stop if error rate exceeds 15%
    errorRate := getErrorRate(checkProps.Ctx)
    if errorRate > 0.15 {
        checkProps.Logger.Warn("Terminating: High error rate",
            zap.Float64("error_rate", errorRate),
        )
        return &models.TerminateRuleActionTune{
            Action: models.ActionStop,
            Reason: "Error rate exceeded 15% threshold",
        }, nil
    }

    // Stop outside business hours if idle for 2+ minutes
    currentHour := time.Now().Hour()
    if currentHour < 9 || currentHour >= 18 {
        idleDuration := time.Since(checkProps.LastMessageAt)
        if idleDuration > 2*time.Minute {
            return &models.TerminateRuleActionTune{
                Action: models.ActionStop,
                Reason: "Outside business hours and idle",
            }, nil
        }
    }

    return &models.TerminateRuleActionTune{
        Action: models.ActionContinue,
    }, nil
}

Real-World Scenarios

Scenario 1: The Smart Streaming Consumer

Challenge: You're consuming from Kafka topics that have variable message rates. During business hours, thousands of messages per second. At night, maybe one every few minutes.

Solution:

func TerminateRule(param *models.TerminateRuleProps) (*models.TerminateRuleTune, error) {
    idleTimeout := 15 * time.Minute
    maxPipelineTime := 6 * time.Hour

    return &models.TerminateRuleTune{
        IdleTimeout:     &idleTimeout,
        MaxPipelineTime: &maxPipelineTime,
        CheckInterval:   30 * time.Second,
    }, nil
}

Result: Pipeline runs as long as data flows. If messages stop for 15 minutes, it shuts down gracefully. You can restart it with a scheduled trigger or when new data arrives. No more 24/7 execution for sporadic workloads.

Scenario 2: The Cost-Conscious Development Pipeline

Challenge: Developers are testing pipeline changes against production-like data. Full dataset has 50 million records, but you only need to verify logic.

Solution:

func TerminateRule(param *models.TerminateRuleProps) (*models.TerminateRuleTune, error) {
    maxRecords := uint64(5000)
    maxPipelineTime := 10 * time.Minute

    return &models.TerminateRuleTune{
        MaxRecords:      &maxRecords,
        MaxPipelineTime: &maxPipelineTime,
        CheckInterval:   5 * time.Second,
    }, nil
}

Result: Each test run processes at most 5,000 records or runs for 10 minutes. Fast feedback, minimal compute costs.

Scenario 3: The Adaptive Business-Hours Pipeline

Challenge: You process user activity events, but your users are primarily in US timezones. Running full capacity globally is wasteful.

Solution:

func TerminateRule(param *models.TerminateRuleProps) (*models.TerminateRuleTune, error) {
    return &models.TerminateRuleTune{
        CheckInterval: 1 * time.Minute,

        UserDefinedCheckFunc: func(checkProps *models.CustomTerminateRuleCheckProps) (*models.TerminateRuleActionTune, error) {
            hour := time.Now().UTC().Hour()

            // US business hours: 14:00-02:00 UTC (9am-9pm EST/PST range)
            isBusinessHours := hour >= 14 || hour < 2

            if !isBusinessHours {
                // Outside business hours, be more aggressive
                idleDuration := time.Since(checkProps.LastMessageAt)
                if idleDuration > 5*time.Minute {
                    return &models.TerminateRuleActionTune{
                        Action: models.ActionStop,
                        Reason: "Outside US business hours with 5min idle",
                    }, nil
                }
            }

            return &models.TerminateRuleActionTune{
                Action: models.ActionContinue,
            }, nil
        },
    }, nil
}

Result: During business hours, pipeline tolerates longer idle periods. Outside business hours, it shuts down quickly if activity drops.

Best Practices: Getting Termination Right

1. Choose the Right CheckInterval

The CheckInterval determines how frequently your termination conditions are evaluated. It's a balance:

Too short (< 5 seconds): Unnecessary overhead checking conditions constantly
Too long (> 1 minute): Slow to respond, pipeline may run longer than needed
Sweet spot: 10–30 seconds for most use cases

// For responsive termination
CheckInterval: 10 * time.Second

// For lower overhead on stable pipelines
CheckInterval: 30 * time.Second

2. Always Log Termination Events

Future you will thank present you for clear termination logging:

checkProps.Logger.Info("Pipeline terminating",
    zap.String("reason", "idle timeout exceeded"),
    zap.Duration("idle_duration", idleDuration),
    zap.Uint64("total_processed", checkProps.TotalMessages),
    zap.Duration("total_runtime", time.Since(checkProps.StartTime)),
)

3. Combine Multiple Conditions

Don't rely on a single termination condition. Defense in depth:

maxRecords := uint64(1000000)
idleTimeout := 10 * time.Minute
maxPipelineTime := 4 * time.Hour

return &models.TerminateRuleTune{
    MaxRecords:      &maxRecords,      // Safety: never process more than 1M
    IdleTimeout:     &idleTimeout,     // Efficiency: stop if quiet
    MaxPipelineTime: &maxPipelineTime, // Control: absolute time limit
    CheckInterval:   20 * time.Second,
}

4. Test in Development First

Termination logic can be tricky. Test with different scenarios:

What happens if no data arrives?
What if data arrives continuously?
What if there are brief idle periods?

Use aggressive limits in development:

// Development settings
maxRecords := uint64(100)
idleTimeout := 1 * time.Minute
maxPipelineTime := 5 * time.Minute

5. Monitor Termination Patterns

Track why your pipelines terminate. If they're always hitting MaxPipelineTime, maybe your processing is too slow. If they're always hitting IdleTimeout, maybe your upstream data source has issues.

The Bottom Line

Termination Rules are your pipeline's exit strategy. They transform potentially runaway processes into well-behaved, cost-effective, and maintainable systems.

In a world where data never stops flowing, knowing when to stop processing is just as important as knowing how to process it. ETLFunnel's Termination Rules give you that control — simply, powerfully, and elegantly.

Your cloud bill will thank you. Your operations team will thank you. And you'll sleep better knowing your pipelines won't run forever.

Ready to optimize your pipeline termination? Visit etlfunnel.com today to sign up for a free trial of our SaaS platform and transform your data engineering workflows.

How to Data Engineer the ETLFunnel Way

Vivek Burman — Sat, 01 Nov 2025 16:19:53 +0000

Part 3 — Checkpoint and Backlog: Building Visibility Into Your Data Pipeline

In Part 1, we covered idempotency, retries, and recovery — the foundational mechanisms that make pipelines resilient to failure.

In Part 2, we explored dynamic orchestration — how to scale ETL workloads intelligently based on available hardware and data volume.

But here's the critical question we haven't answered yet:

When your pipeline runs successfully, how do you track what actually committed?

When it fails, where did the failures go, and how do you replay them?

That's where checkpoint and backlog hooks come in — the twin pillars of pipeline observability and incident management.

The Problem

You've built a robust ETL pipeline. It extracts data from Postgres, transforms it, and loads it into Elasticsearch. Everything works… until you need to answer these questions:

"Which records were successfully committed in the last run?"
"Did the pipeline skip any records during transformation?"
"What happens when a destination write fails?"
"Can I replay failed records without reprocessing the entire dataset?"

Without structured tracking, you're flying blind. Success and failure both happen in a black box. You have logs, sure — but logs are unstructured, hard to query, and don't give you a recovery path.

This is where checkpoint and backlog hooks transform your pipeline from a fire-and-forget script into a fully auditable, recoverable data system.

The Concept: Checkpoint and Backlog Hooks

Checkpoint hooks fire automatically after every successful write to the destination. They answer the question: "What just committed?"

Backlog hooks fire automatically when a write to the destination fails. They answer the question: "What just failed, and how do I handle it?"

Together, they provide:

Data lineage tracking — every committed record is logged with metadata.
Incident management — every failed record is persisted to a structured dead-letter queue (DLQ) with failure context.
Replay capability — backlogged records can be reprocessed as a separate flow, independent of the main pipeline.

Let's see how this works in practice.

The Implementation

Checkpoint Hook: Tracking Success

A checkpoint hook is invoked every time data is successfully written to the destination. It receives:

The committed record(s)
Access to source, destination, and auxiliary databases
The pipeline execution context

Here's a real implementation:

import (
    "etlfunnel/execution/models"
    "etlfunnel/database/cast"
    "encoding/json"
    "time"
)

func Checkpoint(param *models.CheckpointProps) (*models.CheckpointTune, error) {
    if param == nil || param.Ctx == nil || len(param.Records) == 0 {
        return &models.CheckpointTune{Action: models.ActionContinue}, nil
    }

    param.Logger.Info("Checkpoint triggered", zap.Int("record_count", len(param.Records)))

    // Cast auxiliary MySQL connection for audit logging
    mysqlConn, err := cast.CastAsMySQLDBConnection(param.AuxiliaryDBConnMap["mysql"])
    if err != nil {
        param.Logger.Error("Failed to cast MySQL connection", zap.Error(err))
        return nil, err
    }

    // Process each committed record
    for _, record := range param.Records {
        auditEntry := map[string]interface{}{
            "pipeline_name":     param.Ctx.GetName(),
            "record_id":         record["id"],
            "commit_timestamp":  time.Now().UTC(),
            "source_table":      record["_source_table"],
            "destination_table": record["_destination_table"],
            "processing_status": "committed",
        }

        recordJSON, _ := json.Marshal(record)

        // Persist to audit log
        query := `
            INSERT INTO pipeline_audit_log 
            (pipeline_name, record_id, commit_timestamp, record_data, processing_status)
            VALUES (?, ?, ?, ?, ?)
        `

        _, err = mysqlConn.Exec(query,
            auditEntry["pipeline_name"],
            auditEntry["record_id"],
            auditEntry["commit_timestamp"],
            string(recordJSON),
            auditEntry["processing_status"],
        )

        if err != nil {
            param.Logger.Error("Failed to write audit log", zap.Error(err))
            return nil, err
        }
    }

    // Update pipeline statistics
    updatePipelineStats(mysqlConn, param.Ctx.GetName(), len(param.Records))

    return &models.CheckpointTune{Action: models.ActionContinue}, nil
}

func updatePipelineStats(conn *client.Conn, pipelineName string) {
    query := `
        INSERT INTO pipeline_stats (pipeline_name, last_commit_time, record_count)
        VALUES (?, ?, 1)
        ON DUPLICATE KEY UPDATE
        last_commit_time = VALUES(last_commit_time),
        record_count = record_count + 1
    `
    conn.Exec(query, pipelineName, time.Now().UTC())
}

What does this give you?

Full audit trail — every committed record is logged with timestamp, source, destination, and full payload.
Pipeline metrics — track throughput, last commit time, and record count per pipeline.
Event-driven workflows — trigger alerts or downstream processes for specific record types (e.g., high-value transactions).

Backlog Hook: Handling Failure

A backlog hook is invoked when a write to the destination fails. It receives:

The failed record(s)
Access to auxiliary databases for DLQ storage
The pipeline execution context

Here's how you implement it:

import (
    "etlfunnel/execution/models"
    "etlfunnel/database/cast"
    "encoding/json"
    "time"
)

func Backlog(param *models.BacklogProps) (*models.BacklogTune, error) {
    if param == nil || param.Ctx == nil || len(param.Records) == 0 {
        return &models.BacklogTune{Action: models.ActionContinue}, nil
    }

    param.Logger.Error("Write failure detected", zap.Int("failed_record_count", len(param.Records)))

    // Cast auxiliary MySQL connection for DLQ storage
    mysqlConn, err := cast.CastAsMySQLDBConnection(param.AuxiliaryDBConnMap["mysql"])
    if err != nil {
        param.Logger.Error("Failed to cast MySQL connection for backlog", zap.Error(err))
        return nil, err
    }

    // Process each failed record
    for _, record := range param.Records {
        recordJSON, _ := json.Marshal(record)

        // Persist failed record to DLQ
        query := `
            INSERT INTO failed_records 
            (pipeline_name, record_id, record_data, failure_timestamp, retry_count, status)
            VALUES (?, ?, ?, ?, 0, 'pending')
        `

        _, err = mysqlConn.Exec(query,
            param.Ctx.GetName(),
            record["id"],
            string(recordJSON),
            time.Now().UTC(),
        )

        if err != nil {
            param.Logger.Error("Failed to store backlog record", zap.Error(err))
            return nil, err 
        }
    }

    // Update failure statistics
    updateFailureStats(mysqlConn, param.Ctx.GetName(), len(param.Records))

    param.Logger.Info("Records successfully backlogged. Continuing pipeline.", 
        zap.Int("backlogged_count", len(param.Records)))

    return &models.BacklogTune{Action: models.ActionContinue}, nil 
}

func updateFailureStats(conn *client.Conn, pipelineName string, failureCount int) {
    query := `
        INSERT INTO pipeline_failure_stats (pipeline_name, last_failure_time, failure_count)
        VALUES (?, ?, ?)
        ON DUPLICATE KEY UPDATE
        last_failure_time = VALUES(last_failure_time),
        failure_count = failure_count + VALUES(failure_count)
    `
    conn.Exec(query, pipelineName, time.Now().UTC(), failureCount)
}

What does this give you?

Structured DLQ — failed records are stored in a queryable table, not just scattered in logs.
Failure metrics — track failure rates, last failure time, and failure count per pipeline.
Incident triage — identify and prioritize critical failures for immediate attention.
Safe continuation — the main pipeline continues running even when individual writes fail.

The Replay Flow: Turning Failures Into Recovery

Here's where it all comes together.

Failed records in the failed_records table don't just sit there. You can build a separate replay flow that:

Reads records from the DLQ (with status = 'pending')
Applies the same transformations as the main pipeline
Attempts to rewrite to the destination
Updates the DLQ status to 'resolved' or 'failed_again'
Tracks replay progress with its own checkpoints

This is a first-class recovery mechanism — not an ad-hoc script, but a structured flow with the same guarantees as your main pipeline.

Why This Matters

Checkpoint and backlog hooks transform your pipeline from a black box into a glass box.

You get:

Full visibility — every committed record is tracked. Every failed record is captured.
Auditable lineage — trace any record from source to destination, with timestamps and metadata.
Deterministic recovery — replay failed records without reprocessing the entire dataset.
Operational confidence — know exactly what succeeded, what failed, and what's pending.

This is the difference between pipelines that move data and pipelines that move data reliably.

Ready to Build Auditable, Recoverable ETL Pipelines? Visit etlfunnel.com today to sign up for a free trial of our SaaS platform and transform your data engineering workflows.

How to Data Engineer the ETLFunnel Way

Vivek Burman — Sat, 01 Nov 2025 16:14:26 +0000

Part 2 — Dynamic Orchestration

In Part 1, **we discussed **reliability — how idempotency, retries, and recovery make your ETL pipelines resilient.

But once things start scaling, reliability alone isn't enough. You start asking:

"How do I make sure every machine runs at its full potential?"
"What if some datasets are 10× larger than others?"
"Can my orchestration layer adapt automatically?"

That's the heart of today's topic:

Dynamic Orchestration — tuning your ETL workload based on the machine and the data it's running on.

The Problem

You've got 10 flows syncing data from Postgres to Elastic. They're running across multiple regions and machines:

 Region | Machine Type       | Postgres Schemas | Data Volume (per schema)
 -----  | -----------------  | ---------------- | -----------------------
 US     | 32-core, 128GB RAM | 200+             | Medium                   
 EU     | 8-core, 32GB RAM   | 50               | Small                    
 APAC   | 16-core, 64GB RAM  | 100              | Very Large

If you run the same orchestration plan everywhere, you'll waste compute in the US region and choke the EU one.

This is where orchestration hooks come in — to shape your execution plan dynamically before any flow even starts.

The Concept: Orchestration Hooks

An Orchestration Hook is a pre-step that decides how many replicas of a flow or pipeline should exist and how they should be distributed based on:

Available hardware (CPU, cores, memory)
Data volume (number of rows, partitions, tables)
Connection characteristics (source latency, throughput)

You can define hooks at two levels:

Flow Level — to replicate entire flows
Pipeline Level — to replicate pipelines inside a flow

The Implementation

Let's look at how this works in ETLFunnel.

ETLFunnel's engine defines an orchestration contract:

type OrchestratorEntityDef struct {
    Name               string
    SourceDBConn       IDatabaseEngine
    DestDBConn         IDatabaseEngine
    AuxiliaryDBConnMap map[string]IDatabaseEngine
}
type OrchestratorProps struct {
    Entity []OrchestratorEntityDef
}
type OrchestratorTune struct {
    ParentEntityName string
    NewEntityName    string
    ReplicaProps     map[string]any
}

The idea is simple: each entity represents a pipeline unit, and the orchestrator decides how many replicas to create and what their tuning parameters should be.

Flow-level orchestration — scale by CPU

import (
    "etlfunnel/execution/models"
    "fmt"
    "runtime"
)
func GetFlowOrchestration(param *models.OrchestratorProps) ([]models.OrchestratorTune, error) {
    // Dynamically detect available CPU cores
    numThreads := runtime.NumCPU()
    var replicas []models.OrchestratorTune
    for _, entity := range param.Entity {
        for i := 0; i < numThreads; i++ {
            replica := models.OrchestratorTune{
                ParentEntityName: entity.Name,
                NewEntityName:    fmt.Sprintf("%s_core_%d", entity.Name, i),
                ReplicaProps: map[string]any{
                    "replica_id":     i,
                    "thread_id":      i,
                    "total_replicas": numThreads,
                    "cpu_optimized":  true,
                },
            }
            replicas = append(replicas, replica)
        }
    }
    return replicas, nil
}

This is a hardware-aware orchestration. Each pipeline replica maps to a CPU core, so your ETL workload scales automatically with the available cores on that node.

Pipeline-level orchestration — scale by data volume

Let's take it a step further.

You can use your SourceDBConn (which implements IDatabaseEngine) to introspect data size and partition the workload accordingly.

package client_orchestrator_pipeline
import (
    "etlfunnel/execution/models"
    "fmt"
)
func GetPipelineOrchestration(param *models.OrchestratorProps) ([]models.OrchestratorTune, error) {
    if param == nil {
        return nil, fmt.Errorf("orchestrator props cannot be nil")
    }
    var replicas []models.OrchestratorTune
    for _, entity := range param.Entity {
        // Example: fetch table stats using your IDatabaseEngine
        stats, err := entity.SourceDBConn.GetTableStats()
        if err != nil {
            return nil, fmt.Errorf("failed to fetch stats for %s: %v", entity.Name, err)
        }
        // Split pipelines based on total row count
        numReplicas := 1
        if stats.TotalRows > 10_000_000 {
            numReplicas = 4
        } else if stats.TotalRows > 1_000_000 {
            numReplicas = 2
        }
        for i := 0; i < numReplicas; i++ {
            replica := models.OrchestratorTune{
                ParentEntityName: entity.Name,
                NewEntityName:    fmt.Sprintf("%s_partition_%d", entity.Name, i),
                ReplicaProps: map[string]any{
                    "replica_id":     i,
                    "total_replicas": numReplicas,
                    "partition_hint": fmt.Sprintf("split_%d", i),
                    "data_driven":    true,
                },
            }
            replicas = append(replicas, replica)
        }
    }
    return replicas, nil
}

Now orchestration adapts not just to the machine, but also to the data itself. A pipeline that needs to move 100M rows automatically gets partitioned into multiple smaller replicas, each handling a subset.

Combined: Flow + Pipeline Orchestration

In real-world setups, you can even chain both levels.

For example:

Flow orchestration splits jobs by CPU capacity.
Pipeline orchestration further divides heavy tables within each flow.

The result is a perfectly balanced plan — CPU-efficient, data-aware, and regionally optimized.

Why It Matters

Dynamic orchestration gives you:

Predictable performance across heterogeneous environments.
Elastic scalability — every machine operates at full capacity.
Data-aware load balancing — big datasets automatically spread across workers.
Unified code — your flow definitions remain the same, only orchestration changes.

You've effectively made your ETL system self-tuning — a foundational step toward distributed dataflow intelligence.

Ready to build resilient, idempotent ETL pipelines that handle can scale to any environment? Visit etlfunnel.com today to sign up for a free trial of our SaaS platform and transform your data engineering workflows.

How to Data Engineer the ETLFunnel Way

Vivek Burman — Sat, 01 Nov 2025 16:07:49 +0000

Part 1 — Idempotency, Retry, and Recovery

Modern data engineering isn't about moving data from A to B — it's about doing it correctly every single time. The deeper a pipeline goes, the more fragile it becomes. A single failed batch can leave datasets half-written, cause duplicates downstream, or silently skip records. The hardest part isn't transformation logic — it's keeping the system consistent when things go wrong. At ETLFunnel, we solve these challenges to ensure reliable, scalable data pipelines for developers and teams.

The Problem: When Retry Becomes Corruption

A failed task is easy to rerun, but doing so safely isn't.
If a data step partially commits results before crashing, the next retry might reprocess the same data again — duplicating rows or overwriting the wrong partitions. Traditional batch pipelines (like simple cron + SQL scripts) have no built-in concept of idempotency — they only know how to rerun.

Retries without idempotency create corruption.
Retries without structured state tracking create blind spots.
Retries without visibility create operational debt.

Dead-letter queues (DLQs) are supposed to help here, but most implementations are crude: a folder of failed records, no schema, no context, no replay mechanism. They collect data but don't provide an actual recovery path.

Data engineers end up writing retry scripts, marking rows manually, or purging partial results — a process that doesn't scale or audit well.

The Solution: Declarative Checkpoints and Backlogs

The fix begins with a change in mindset: a pipeline must remember where it was and record what failed — consistently.

Instead of treating idempotency as a postmortem patch, we treat it as a first-class citizen of execution. Each step carries three key definitions:

Checkpoint Hook— The checkpoint hook receives a *models.CheckpointProps. Read the current context and records from that struct, resolve an auxiliary DB engine via CastAsPostgresDBConnection(...), and persist a compact checkpoint row (pipeline identifier + last-processed id + timestamp). The hook returns a *models.CheckpointTune indicating whether the pipeline should continue or stop. The example below shows a conservative, idempotent upsert into a pipeline_checkpoints table and logs useful events via the provided ILoggerContract.

import (
 "context"
 "fmt"
 "time"

 "etlfunnel/execution/models"

 "github.com/jackc/pgx/v5"
)

// Checkpoint persists last processed ID to Postgres.
func Checkpoint(param *models.CheckpointProps) (*models.CheckpointTune, error) {
 if param == nil || param.Ctx == nil || len(param.Records) == 0 {
  return &models.CheckpointTune{Action: models.ActionContinue}, nil
 }

 ctx := param.Ctx.GetContext()
 logger := param.Logger

 // Extract last ID
 lastRec := param.Records[len(param.Records)-1]
 lastIDRaw, ok := lastRec["id"]
 if !ok {
  logger.Warn("No 'id' field; skipping persist")
  return &models.CheckpointTune{Action: models.ActionContinue}, nil
 }
 lastIDStr := fmt.Sprintf("%v", lastIDRaw)

 // Resolve Postgres connection (auxiliary or fallback to dest)
 var engine models.IDatabaseEngine
 if param.AuxiliaryDBConnMap != nil {
  if e, ok := param.AuxiliaryDBConnMap["checkpoint_db"]; ok {
   engine = e
  }
 }
 if engine == nil {
  engine = param.DestDBConn
 }
 pgConn, err := models.CastAsPostgresDBConnection(engine)
 if err != nil {
  return &models.CheckpointTune{Action: models.ActionStop}, fmt.Errorf("cast failed: %w", err)
 }

 // Ensure table and upsert
 createSQL := `CREATE TABLE IF NOT EXISTS pipeline_checkpoints (
  pipeline_name TEXT PRIMARY KEY, last_id TEXT, updated_at TIMESTAMP WITH TIME ZONE
 )`
 pgConn.Exec(ctx, createSQL)

 pipelineName := getPipelineName(param.Ctx) // Helper to derive name
 upsertSQL := `INSERT INTO pipeline_checkpoints (pipeline_name, last_id, updated_at)
  VALUES ($1, $2, $3) ON CONFLICT (pipeline_name) DO UPDATE SET last_id = EXCLUDED.last_id, updated_at = EXCLUDED.updated_at`
 if _, err := pgConn.Exec(ctx, upsertSQL, pipelineName, lastIDStr, time.Now().UTC()); err != nil {
  logger.Error("Persist failed")
  return &models.CheckpointTune{Action: models.ActionStop}, err
 }

 logger.Info(fmt.Sprintf("Checkpoint for %s at %s", pipelineName, lastIDStr))
 return &models.CheckpointTune{Action: models.ActionContinue}, nil
}

The checkpoint commits after a successful batch, ensuring that the next retry begins exactly from the last consistent point.

Backlog Function — following the same pattern as the checkpoint function. It uses the *models.BacklogProps struct, persists failed records to an auxiliary Postgres DB, and returns a *models.BacklogTune indicating whether to continue or stop the pipeline.

import (
 "context"
 "encoding/json"
 "fmt"
 "time"

 "etlfunnel/execution/models"

 "github.com/jackc/pgx/v5"
)

// Backlog persists failed records to DLQ table.
func Backlog(param *models.BacklogProps) (*models.BacklogTune, error) {
 if param == nil || param.Ctx == nil || len(param.Records) == 0 {
  return &models.BacklogTune{Action: models.ActionContinue}, nil
 }

 ctx := param.Ctx.GetContext()
 logger := param.Logger

 // Resolve Postgres connection
 var engine models.IDatabaseEngine
 if param.AuxiliaryDBConnMap != nil {
  if e, ok := param.AuxiliaryDBConnMap["backlog_db"]; ok {
   engine = e
  }
 }
 if engine == nil {
  engine = param.DestDBConn
 }
 pgConn, err := models.CastAsPostgresDBConnection(engine)
 if err != nil {
  return &models.BacklogTune{Action: models.ActionStop}, fmt.Errorf("cast failed: %w", err)
 }

 // Ensure DLQ table
 createSQL := `CREATE TABLE IF NOT EXISTS pipeline_backlog (
  id SERIAL PRIMARY KEY, pipeline_name TEXT, record_data JSONB, error_time TIMESTAMP WITH TIME ZONE DEFAULT now()
 )`
 pgConn.Exec(ctx, createSQL)

 pipelineName := getPipelineName(param.Ctx)
 for _, rec := range param.Records {
  dataBytes, err := json.Marshal(rec)
  if err != nil {
   logger.Warn(fmt.Sprintf("Marshal failed: %v", err))
   continue
  }

  insertSQL := `INSERT INTO pipeline_backlog (pipeline_name, record_data, error_time) VALUES ($1, $2, $3)`
  if _, err := pgConn.Exec(ctx, insertSQL, pipelineName, dataBytes, time.Now().UTC()); err != nil {
   logger.Error("Insert failed")
   return &models.BacklogTune{Action: models.ActionStop}, err
  }
 }

 logger.Info(fmt.Sprintf("Backlog: %d records for %s", len(param.Records), pipelineName))
 return &models.BacklogTune{Action: models.ActionContinue}, nil
}

Each backlog entry includes structured metadata: the step ID, failure reason, and serialized payload. Failed records are never retried automatically in the main pipeline. Instead, they form the source for separate replay flows.

Separate Replay Flow

Once records are persisted in the backlog, they can be processed as a dedicated flow:

This replay flow reads records from the DLQ or auxiliary database.
It applies the same transformations and business logic as the original pipeline.
It can also include additional validation, sampling, or recovery logic.
Progress in the replay flow is tracked independently using its own checkpoints, providing auditable, deterministic recovery.

By separating normal execution from replay, you avoid mixing failure handling with live processing, simplify debugging, and maintain idempotent guarantees across both flows. ETLFunnel automates this replay capability to make recovery seamless for on-premise and cloud deployments.

Why This Works

Idempotency by Design: Checkpoints persist logical progress, not just job status.
Structured Recovery: Backlogs store typed errors, making them queryable and auditable.
Replay as a First-Class Flow: Recovery is explicit and traceable, avoiding accidental reprocessing in the main pipeline.
Reviewable Operations: Failures aren't opaque; they're captured with metadata and visible through structured logs or UI.

This approach ensures that every record — successful or failed — has a deterministic, auditable path, while keeping the main pipeline clean and focused.

Ready to build resilient, idempotent ETL pipelines that handle failures gracefully? Visit etlfunnel.com today to sign up for a free trial of our SaaS platform and transform your data engineering workflows.

This is TESt

Vivek Burman — Tue, 02 Jun 2020 16:36:56 +0000

I am a big proponent of always thinking about scale and optimization when you are writing code, no matter how big or small of a product you are working on. A lot of people like to argue that there is no reason to optimize when you don't have to and that is wastes time. Yes, sometimes it does take a little bit of extra time, but down the road that little bit of time can save you a huge headache.

I'm not talking about rearchitecting a whole feature, I'm talking about the little things you can do while you are coding that I guarantee will make a difference in the future. In this post, I will walk through an example of a small optimization I made recently at DEV with this pull request and how we could of avoided creating that pull request altogether.

The Most Popular DEV endpoint