Overview

Celery's countdown and eta arguments map directly onto KubeMQ's native delay_in_seconds on the Queues messages that carry every task. The broker holds the message server-side and delivers it when the delay elapses — there is zero client-side polling and no broker plugin to install.

This is a real difference from other brokers. With Redis, Celery implements countdown/eta using a client-side polling loop. With RabbitMQ, you need the rabbitmq_delayed_message_exchange plugin. With KubeMQ, delayed delivery is a built-in property of the queue message:

Broker	Delayed delivery mechanism
Redis	Client-side polling loop
RabbitMQ	`rabbitmq_delayed_message_exchange` plugin
KubeMQ	Native `delay_in_seconds` — no polling, no plugin

From the API perspective nothing changes — apply_async(countdown=...) and apply_async(eta=...) work exactly as they do on any Celery broker:

task.apply_async(countdown=60)          # delivers after 60 seconds
task.apply_async(eta=future_datetime)   # delivers at a specific time

One-time scheduling vs. recurring schedules. countdown and eta schedule a task to run once at a future time. For tasks that repeat on a fixed cadence (every minute, daily, at sunset), use Celery Beat — covered later on this page.

Run a broker

All examples on this page connect to a KubeMQ broker over the native gRPC port 50000. The Celery transport is a Kombu virtual transport — it talks gRPC directly, so no HTTP connector flag is required (port 9090 is the shared HTTP server for the CloudEvents, REST, MCP, and A2A connectors, which Celery does not use).

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

pip install kubemq-celery

Countdown

apply_async(countdown=N) tells KubeMQ to hold the message for N seconds before delivering it to a worker. Tasks with a shorter countdown are delivered first, regardless of dispatch order.

countdown_delay.py

import os
import time

from celery import Celery

import kubemq_celery  # noqa: F401

app = Celery("countdown_delay")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
    }
)


@app.task
def delayed_greeting(name: str) -> dict:
    """A greeting that arrives after a countdown delay."""
    return {"message": f"Hello, {name}!", "delivered_at": time.time()}


@app.task
def scheduled_cleanup(resource_id: str) -> dict:
    """Clean up a resource after a delay."""
    return {"resource_id": resource_id, "action": "cleaned_up", "at": time.time()}

Dispatch tasks with different countdowns. countdown=N maps to KubeMQ's delay_in_seconds — no client-side polling is involved:

# Delivered 5s after dispatch.
r1 = delayed_greeting.apply_async(args=("Alice",), countdown=5)

# Dispatched second, but delivered first (shorter delay).
r2 = delayed_greeting.apply_async(args=("Bob",), countdown=3)

# Delivered 10s after dispatch.
r3 = scheduled_cleanup.apply_async(args=("res-001",), countdown=10)

Run a worker against the broker to observe the delays:

celery -A countdown_delay worker --loglevel=info

ETA

apply_async(eta=datetime) schedules a task for an exact wall-clock time. The transport computes delay_in_seconds = (eta - now).total_seconds() and hands that to KubeMQ. Always pass a timezone-aware datetime — UTC is recommended, and Celery's enable_utc keeps ETA math unambiguous.

eta_scheduling.py

import os
import time
from datetime import datetime, timedelta, timezone

from celery import Celery

import kubemq_celery  # noqa: F401

app = Celery("eta_scheduling")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "enable_utc": True,
    }
)


@app.task
def run_at_time(label: str) -> dict:
    """Task scheduled to run at a specific ETA."""
    return {
        "label": label,
        "executed_at": datetime.now(timezone.utc).isoformat(),
        "timestamp": time.time(),
    }

Schedule tasks for specific future times. As with countdown, the task with the earlier ETA is delivered first:

now = datetime.now(timezone.utc)

# Runs 10 seconds from now.
run_at_time.apply_async(args=("10s-from-now",), eta=now + timedelta(seconds=10))

# Scheduled after the first call, but executes earlier (earlier ETA).
run_at_time.apply_async(args=("5s-from-now",), eta=now + timedelta(seconds=5))

eta vs. countdown — eta=datetime runs at a specific timezone-aware time; countdown=seconds runs after N seconds from now. Both compile down to the same KubeMQ delay_in_seconds primitive. Use timezone-aware datetimes (UTC recommended) to avoid drift.

The 24-hour delay cap

KubeMQ's delay_in_seconds caps at 86400 seconds (24 hours). If a countdown or eta exceeds 24 hours, the transport caps the effective delay at 24h and emits a warning log rather than rejecting the task. For longer horizons, use Celery Beat (below).

Delays over 24 hours are capped. A countdown=172800 (48h) is delivered after 24h, not 48h, with a warning. The same cap applies to ETA values more than 24 hours in the future, and to result_expires (queue message expiration, also 86400s max). For schedules beyond a day, use Celery Beat; for one-shot tasks beyond a day, store the target time in a database and let Beat poll it.

The behavior is illustrated below — delays at or under the limit pass through, while anything over is clamped to 86400:

beat_max_delay_warning.py

MAX_DELAY = 86400  # 24 hours in seconds

delay_tests = [
    ("within-limit-1h", 3600),    # OK — delivered after 1 hour
    ("within-limit-12h", 43200),  # OK — delivered after 12 hours
    ("at-limit-24h", 86400),      # OK — delivered after exactly 24 hours
    ("over-limit-48h", 172800),   # CAPPED — 48h requested, delivered after 24h
    ("over-limit-7d", 604800),    # CAPPED — 7 days requested, delivered after 24h
]

for label, delay in delay_tests:
    effective = min(delay, MAX_DELAY)
    if delay > MAX_DELAY:
        print(f"WARNING: delay {delay}s capped to {effective}s")

Celery Beat — crontab schedules

For recurring tasks, run the Celery Beat scheduler. Beat owns the schedule and publishes each task to KubeMQ when it is due; workers then pick it up like any other message. Use crontab() for clock-based cadences (specific minutes, hours, days of week, days of month).

beat_crontab.py

import os
from datetime import datetime, timezone

from celery import Celery
from celery.schedules import crontab

import kubemq_celery  # noqa: F401

app = Celery("beat_crontab")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "enable_utc": True,
        "beat_schedule": {
            # Every minute (crontab() defaults to every minute).
            "every-minute-health": {
                "task": "beat_crontab.health_check",
                "schedule": crontab(),
            },
            # Every 5 minutes.
            "every-5min-metrics": {
                "task": "beat_crontab.collect_metrics",
                "schedule": crontab(minute="*/5"),
            },
            # Daily at midnight UTC.
            "daily-report": {
                "task": "beat_crontab.generate_daily_report",
                "schedule": crontab(minute=0, hour=0),
            },
            # Every Monday at 9:00 AM UTC.
            "weekly-digest": {
                "task": "beat_crontab.send_weekly_digest",
                "schedule": crontab(minute=0, hour=9, day_of_week="monday"),
            },
            # 1st of every month at 6:00 AM UTC.
            "monthly-billing": {
                "task": "beat_crontab.run_monthly_billing",
                "schedule": crontab(minute=0, hour=6, day_of_month=1),
            },
        },
    }
)


@app.task
def health_check() -> dict:
    return {"status": "healthy", "at": datetime.now(timezone.utc).isoformat()}

Start a worker and a Beat scheduler in separate processes:

# Terminal 1: worker
celery -A beat_crontab worker --loglevel=info

# Terminal 2: beat scheduler
celery -A beat_crontab beat --loglevel=info

Run exactly one Beat instance. Beat state is local — it lives in a celerybeat-schedule file, not in KubeMQ. Running multiple Beat processes produces duplicate task dispatches. The combined worker -B form is for development only; in production, run a single dedicated Beat process.

Celery Beat — intervals and solar schedules

Beyond crontab, Beat supports fixed-interval schedules with timedelta and astronomical schedules with solar.

A timedelta schedule runs a task at a fixed interval measured from its previous run. Use it for heartbeats, polling, and periodic syncs. Entries can carry positional args and keyword kwargs.

beat_timedelta.py

from datetime import timedelta

# inside app.config_from_object({... "beat_schedule": { ... }})
"beat_schedule": {
    # Every 10 seconds.
    "heartbeat-10s": {
        "task": "beat_timedelta.heartbeat",
        "schedule": timedelta(seconds=10),
    },
    # Every 30 seconds, with positional args.
    "sensor-reading-30s": {
        "task": "beat_timedelta.read_sensor",
        "schedule": timedelta(seconds=30),
        "args": ("temperature",),
    },
    # Every 2 minutes, with keyword args.
    "cache-refresh-2m": {
        "task": "beat_timedelta.refresh_cache",
        "schedule": timedelta(minutes=2),
        "kwargs": {"cache_name": "user_profiles", "ttl": 300},
    },
    # Every hour.
    "hourly-sync": {
        "task": "beat_timedelta.sync_data",
        "schedule": timedelta(hours=1),
    },
}

timedelta(seconds=30) fires at a fixed 30-second interval from the last run, whereas crontab(minute="*/1") fires at specific clock times. Choose timedelta for "every N seconds/minutes" cadences and crontab for "at these clock times" cadences.

A solar schedule fires relative to a solar event (sunrise, sunset, dawn, dusk) at a given latitude/longitude. Solar schedules recompute the next event time dynamically, accounting for daylight saving time and seasonal change. They require the ephem package (pip install ephem).

beat_solar.py

from celery.schedules import solar

NYC_LAT = 40.7128
NYC_LON = -74.0060

# inside the beat_schedule mapping
"beat_schedule": {
    "sunset-lights-on": {
        "task": "beat_solar.control_lights",
        "schedule": solar("sunset", NYC_LAT, NYC_LON),
        "kwargs": {"action": "on", "zone": "outdoor"},
    },
    "sunrise-lights-off": {
        "task": "beat_solar.control_lights",
        "schedule": solar("sunrise", NYC_LAT, NYC_LON),
        "kwargs": {"action": "off", "zone": "outdoor"},
    },
    "dawn-data-collection": {
        "task": "beat_solar.start_data_collection",
        "schedule": solar("dawn_astronomical", NYC_LAT, NYC_LON),
    },
    "dusk-data-stop": {
        "task": "beat_solar.stop_data_collection",
        "schedule": solar("dusk_astronomical", NYC_LAT, NYC_LON),
    },
}

Available solar events include dawn_astronomical, dawn_nautical, dawn_civil, sunrise, solar_noon, sunset, dusk_civil, dusk_nautical, and dusk_astronomical.

Dynamic and DB-backed schedules

app.conf.beat_schedule is a plain dictionary, so you can add, modify, and remove entries at runtime. This works when Beat runs in the same process as the change (for example, a worker started with the -B flag).

dynamic_periodic_tasks.py

from datetime import timedelta


def add_periodic_task(name, task, schedule, args=None, kwargs=None):
    """Add a periodic task to the beat schedule at runtime."""
    entry = {"task": task, "schedule": schedule}
    if args:
        entry["args"] = args
    if kwargs:
        entry["kwargs"] = kwargs
    app.conf.beat_schedule[name] = entry


def remove_periodic_task(name):
    """Remove a periodic task from the beat schedule."""
    app.conf.beat_schedule.pop(name, None)


# Add a task that runs every 15 seconds.
add_periodic_task(
    "monitor-api",
    "dynamic_periodic_tasks.monitor_service",
    timedelta(seconds=15),
    args=("api-gateway",),
)

# Modify an existing entry's interval.
app.conf.beat_schedule["initial-heartbeat"]["schedule"] = timedelta(seconds=60)

# Remove an entry.
remove_periodic_task("monitor-api")

In-memory changes are lost on restart. The default scheduler keeps state in a local file, so runtime edits to beat_schedule do not survive a Beat restart. For schedules that must persist and be editable without code changes, install django-celery-beat (a DB-backed scheduler) — it ships in this project's examples extra:

pip install "kubemq-celery[examples]"   # includes django-celery-beat >= 2.5

One-shot scheduled tasks

A common pattern is scheduling a single task to fire once at a future moment — sending a confirmation email, expiring an offer, releasing a held reservation. Use eta for an absolute time or countdown for a relative delay. These are one-time dispatches, distinct from Beat's recurring entries.

one_shot_scheduled.py

from datetime import datetime, timedelta, timezone

now = datetime.now(timezone.utc)

# Fire once, 10 seconds from now, at an absolute time.
send_email.apply_async(
    kwargs={
        "to": "user@example.com",
        "subject": "Your order confirmation",
        "body": "Thank you for your purchase!",
    },
    eta=now + timedelta(seconds=10),
)

# Fire once, after a relative delay.
expire_offer.apply_async(args=("PROMO-2024-SPRING",), countdown=15)

# Fire once, at an absolute release time.
release_reservation.apply_async(args=("RES-12345",), eta=now + timedelta(seconds=20))

Both forms use KubeMQ's native delay_in_seconds and are subject to the 24-hour cap. For a one-shot task more than 24 hours out, persist the target time in a database and let a Beat task poll for due items.

Beat in production

In production, run Beat as a dedicated scheduler container — separate from your workers and with exactly one replica. The project ships a Docker Compose file that wires up a KubeMQ broker, two workers on different queues, and a single Beat scheduler:

docker-compose.yaml

services:
  kubemq:
    image: europe-docker.pkg.dev/kubemq/images/kubemq:next
    ports:
      - "50000:50000"  # gRPC
      - "9090:9090"    # Shared HTTP server (REST/health)
      - "8080:8080"    # Management API dashboard
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/health"]
      interval: 10s
      timeout: 5s
      retries: 3
      start_period: 10s

  worker-default:
    build:
      context: ../..
      dockerfile: examples/kubernetes/Dockerfile
    command: celery -A basic_task worker --loglevel=info -Q celery --concurrency=2
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

  # Dedicated Beat scheduler — exactly one instance.
  beat:
    build:
      context: ../..
      dockerfile: examples/kubernetes/Dockerfile
    command: celery -A basic_task beat --loglevel=info
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

cd examples/kubernetes
docker compose up -d

Workers and Beat share the same broker but run as independent services, so you can scale workers freely while keeping a single Beat process — the rule that avoids duplicate dispatches. For a full Kubernetes deployment with service discovery and KEDA autoscaling, see the Kubernetes deployment guide.

Scheduling & Delayed Delivery

Overview

Run a broker

Countdown

ETA

The 24-hour delay cap

Celery Beat — crontab schedules

Celery Beat — intervals and solar schedules

Dynamic and DB-backed schedules

One-shot scheduled tasks

Beat in production

Next steps

Canvas Workflows

Configuration

Kubernetes Deployment

On this page