Duplicate LangChain pipeline runs after overlapping CronJob schedules

Duplicate LangChain Pipeline Runs After Overlapping CronJob Schedules

Problem

In a production deployment, a LangChain agent consumes a high‑volume event stream (Kafka, Kinesis, etc.) and processes each event through a pipeline that writes results to a downstream store. The pipeline is triggered by a Kubernetes CronJob that runs every 5 minutes. After a recent schedule change, operators observed that the same batch of events was processed twice, leading to duplicate database rows, inflated metrics, and downstream idempotency failures.

Typical symptoms:

• Log entries showing two separate pod executions with identical run_id values.
• Database primary‑key violations or “duplicate key” errors.
• Metrics spikes: pipeline_success_total doubled for the affected interval.
• Alert “Duplicate processing detected” firing on the monitoring dashboard.

Root Cause

The CronJob definition was modified to change the schedule from */5 * * * * to 0,5,10,15,20,25,30,35,40,45,50,55 * * * * to align with a new upstream data window. However, the startingDeadlineSeconds field was left at its default (null), and the concurrencyPolicy remained set to Allow. When a previous run took longer than 5 minutes (common during peak traffic), the next scheduled start occurred while the earlier pod was still active. Because Allow permits concurrent executions, the scheduler launched a second pod, both of which read the same offset range from the streaming source (the source does not commit offsets until the pipeline finishes). Consequently, the two pods processed the identical event batch.

Key contributing factors:

• Concurrency policy: Allow instead of Forbid or Replace.
• Missing startingDeadlineSeconds: No deadline caused the scheduler to consider a missed start as a new start rather than a retry.
• Non‑idempotent offset handling: The pipeline relied on exclusive execution rather than explicit offset commits.

Debug

Step‑by‑step investigation performed in the affected namespace:

1. Inspect the CronJob definition.
2. Correlate pod start times with log timestamps.
3. Examine the streaming source offset commits.
4. Capture a short packet trace to verify that both pods subscribed to the same partition.

Commands used:

# 1. Dump the CronJob manifest
kubectl get cronjob langchain-pipeline -n prod -o yaml

# 2. List recent CronJob jobs and their start times
kubectl get jobs -l job-name=langchain-pipeline -n prod --sort-by=.status.startTime

# 3. Show logs for the two overlapping pods
kubectl logs $(kubectl get pods -l job-name=langchain-pipeline-abc123 -n prod -o name) -n prod
kubectl logs $(kubectl get pods -l job-name=langchain-pipeline-def456 -n prod -o name) -n prod

Sample log excerpt (both pods share the same run_id):

2026-09-26T12:00:02.123Z INFO run_id=202609261200 pipeline_start processing batch offset=15000-16000
2026-09-26T12:00:02.127Z INFO run_id=202609261200 pipeline_start processing batch offset=15000-16000

Offset commit check (Kafka consumer group):

# Verify committed offsets for the consumer group used by the pipeline
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group langchain-pipeline-group

Output showed no offset advancement for the overlapping interval, confirming both pods read the same range.

Solution

Two complementary changes were applied:

1. Update CronJob Concurrency Policy

Set concurrencyPolicy to Forbid so that a new run is blocked while a previous one is still executing.

# Before
apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: langchain-pipeline
spec:
  schedule: "0,5,10,15,20,25,30,35,40,45,50,55 * * * *"
  concurrencyPolicy: Allow
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: pipeline
            image: myrepo/langchain-pipeline:1.2.3

# After
apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: langchain-pipeline
spec:
  schedule: "0,5,10,15,20,25,30,35,40,45,50,55 * * * *"
  concurrencyPolicy: Forbid      # ← changed
  startingDeadlineSeconds: 300   # optional safety net
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: pipeline
            image: myrepo/langchain-pipeline:1.2.3

2. Make Offset Commit Explicit and Idempotent

Modify the pipeline code to commit offsets after each successful batch and to store a deterministic run_id derived from the batch’s start offset. This ensures that even if a duplicate pod starts, it will detect that the offsets have already been processed.

# langchain/pipeline.py (excerpt)
def process_batch(batch):
    run_id = f"{batch.start_offset}-{batch.end_offset}"
    if already_processed(run_id):
        logger.info("Skipping already processed batch %s", run_id)
        return

    # ... processing logic ...

    commit_offsets(batch.end_offset)
    mark_processed(run_id)

Helper functions:

def already_processed(run_id):
    return redis_client.sismember("processed_batches", run_id)

def mark_processed(run_id):
    redis_client.sadd("processed_batches", run_id)

Verify

After applying the changes, the following checks confirm the issue is resolved:

1. Observe that only one pod is created per schedule.
2. Confirm that the run_id appears once in the processed_batches set.
3. Check Kafka consumer group offsets have advanced after each run.
4. Validate that no duplicate primary‑key errors appear in the downstream database.

Verification commands:

# Ensure only one job per schedule
kubectl get jobs -l job-name=langchain-pipeline -n prod --field-selector=status.successful=1

# Verify processed_batches set size grows monotonically
redis-cli SMEMBERS processed_batches | wc -l

# Confirm offset advancement
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group langchain-pipeline-group

Metrics after the fix show a single pipeline_success_total increment per interval, and alerts for duplicate processing are silent.

Prevent

• Enforce concurrencyPolicy: Forbid on all time‑driven CronJob workloads that are not explicitly designed for parallelism.
• Set startingDeadlineSeconds to a sensible value (e.g., 300 s) to avoid runaway retries when a run is missed.
• Make pipelines idempotent by persisting a deterministic batch identifier and checking it before processing.
• Monitor job duration and set alerts when the runtime exceeds a configurable threshold (e.g., 80 % of the schedule interval).
• Automated CI linting for Kubernetes manifests to flag missing or unsafe concurrencyPolicy settings.

FAQ

1. Why did the duplicate runs only appear after the schedule change?

   The new schedule introduced a 5‑minute interval that matched the average runtime of the pipeline during peak load. The previous schedule (every 10 minutes) gave enough headroom for the job to finish before the next trigger.

2. Can I keep concurrencyPolicy: Allow and still avoid duplicates?

   Yes, but you must implement explicit locking (e.g., a distributed mutex in Redis or etcd) and ensure offset commits are idempotent. This adds complexity and is usually unnecessary for batch‑style pipelines.

3. What is the purpose of startingDeadlineSeconds?

   If a scheduled start is missed (e.g., due to node outage), the scheduler will attempt to start the job within the deadline. Setting it prevents indefinite retries that could overlap with future runs.

4. How do I detect that a job is taking longer than expected?

   Expose the pod’s runtime as a Prometheus gauge (e.g., pipeline_job_duration_seconds) and create an alert when it exceeds 0.8 * schedule_interval. Additionally, enable activeDeadlineSeconds on the job spec to enforce a hard timeout.

5. Is there a way to automatically back‑fill missed offsets if a run fails?

   Implement a “reconciliation” task that scans the processed_batches set, compares it to the latest offset in the stream, and re‑processes any gaps. This task should also respect the same idempotency guarantees.

Related Topic Hub: RAG Systems Troubleshooting Hub