pathwaycom/pathway · error · ValueError

'start_from_timestamp_ms' must be non-negative; got {start_f

Error message

'start_from_timestamp_ms' must be non-negative; got {start_from_timestamp_ms}. The value is a Unix timestamp in milliseconds — negative values are pre-epoch and not meaningful for Kafka.

What it means

start_from_timestamp_ms in pw.io.kafka.read is a Unix timestamp in milliseconds that tells the reader to seek to the first message at or after that time. Negative values would be pre-1970, which is meaningless for Kafka offsets, so Pathway rejects them with ValueError before the connector starts.

Source

Thrown at python/pathway/io/kafka/__init__.py:275

        raise ValueError(
            "rdkafka_settings must contain a non-empty 'group.id' entry: "
            "Pathway's Kafka reader uses 'subscribe' (not 'assign'), which "
            "librdkafka refuses to perform without a configured consumer "
            f"group id; got {rdkafka_settings.get('group.id')!r}."
        )

    if max_backlog_size is not None and max_backlog_size <= 0:
        raise ValueError(
            f"'max_backlog_size' must be positive; got {max_backlog_size}. "
            f"A non-positive value would prevent any entry from being "
            f"processed and the reader would never make progress."
        )
    if parallel_readers is not None and parallel_readers <= 0:
        raise ValueError(
            f"'parallel_readers' must be positive; got {parallel_readers}."
        )
    if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:
        raise ValueError(
            f"'start_from_timestamp_ms' must be non-negative; got "
            f"{start_from_timestamp_ms}. The value is a Unix timestamp in "
            f"milliseconds — negative values are pre-epoch and not "
            f"meaningful for Kafka."
        )
    if autocommit_duration_ms is not None and autocommit_duration_ms <= 0:
        raise ValueError(
            f"'autocommit_duration_ms' must be positive; got "
            f"{autocommit_duration_ms}. It is the maximum time between "
            f"two commits and zero/negative values would prevent commits "
            f"from happening."
        )

    # When 'start_from_timestamp_ms' is set, the engine seeks lazily after
    # the consumer is positioned at the partition's earliest offset, so any
    # user-supplied 'auto.offset.reset' value that doesn't already mean
    # "start at the beginning" is silently rewritten on the Rust side.
    # Surface that rewrite explicitly so somebody who picked 'latest' on

View on GitHub (pinned to fa2f74a464)

Solutions

1. Pass a Unix timestamp in milliseconds: start_from_timestamp_ms=int(time.time() * 1000) - 3600_000 for one hour ago.
2. Use None (or omit) when you don't want timestamp-based seeking — do not use -1 as a sentinel.
3. Double-check units: Kafka wants milliseconds, not seconds.

Example fix

# before
import time
t = pw.io.kafka.read(rdkafka_settings, topic="t", start_from_timestamp_ms=-1)
# after
import time
t = pw.io.kafka.read(
    rdkafka_settings,
    topic="t",
    start_from_timestamp_ms=int(time.time() * 1000) - 3_600_000,
)

if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:
    raise SystemExit("start_from_timestamp_ms must be a non-negative Unix ms timestamp")

# helper: build safely from a datetime
import time
def ms_ago(seconds: int) -> int:
    return max(0, int(time.time() * 1000) - seconds * 1000)

def valid_start_ts(v) -> bool:
    return v is None or (isinstance(v, int) and v >= 0)

When it happens

Trigger: pw.io.kafka.read(..., start_from_timestamp_ms=-1) or any negative number, including accidentally passing seconds instead of milliseconds with a negative correction, or passing -1 as a sentinel 'not set' value instead of None.

Common situations: Using -1 or 0-adjacent sentinels from other APIs; computing now_ms() - offset where the offset exceeds the current time (e.g. misinterpreted units producing a huge subtraction); passing a datetime in seconds (e.g. int(time.time()) instead of int(time.time()*1000)) then subtracting.

Related errors

• Failed to install dependencies
• Column {pseudocolumn} has to contain integers only.
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times
• Unsupported argument for {data_format!r} format: 'value'
• 'delimiter' is only meaningful for the 'dsv' format, but {fo

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/9ea87c4cc84a1956. Report an issue: GitHub.