{"record":{"id":"9ea87c4cc84a1956","repo":"pathwaycom/pathway","slug":"start-from-timestamp-ms-must-be-non-negative-go","errorCode":null,"errorMessage":"'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.","messagePattern":"'start_from_timestamp_ms' must be non-negative; got (.+?)\\. The value is a Unix timestamp in milliseconds — negative values are pre-epoch and not meaningful for Kafka\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":275,"sourceCode":"        raise ValueError(\n            \"rdkafka_settings must contain a non-empty 'group.id' entry: \"\n            \"Pathway's Kafka reader uses 'subscribe' (not 'assign'), which \"\n            \"librdkafka refuses to perform without a configured consumer \"\n            f\"group id; got {rdkafka_settings.get('group.id')!r}.\"\n        )\n\n    if max_backlog_size is not None and max_backlog_size <= 0:\n        raise ValueError(\n            f\"'max_backlog_size' must be positive; got {max_backlog_size}. \"\n            f\"A non-positive value would prevent any entry from being \"\n            f\"processed and the reader would never make progress.\"\n        )\n    if parallel_readers is not None and parallel_readers <= 0:\n        raise ValueError(\n            f\"'parallel_readers' must be positive; got {parallel_readers}.\"\n        )\n    if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:\n        raise ValueError(\n            f\"'start_from_timestamp_ms' must be non-negative; got \"\n            f\"{start_from_timestamp_ms}. The value is a Unix timestamp in \"\n            f\"milliseconds — negative values are pre-epoch and not \"\n            f\"meaningful for Kafka.\"\n        )\n    if autocommit_duration_ms is not None and autocommit_duration_ms <= 0:\n        raise ValueError(\n            f\"'autocommit_duration_ms' must be positive; got \"\n            f\"{autocommit_duration_ms}. It is the maximum time between \"\n            f\"two commits and zero/negative values would prevent commits \"\n            f\"from happening.\"\n        )\n\n    # When 'start_from_timestamp_ms' is set, the engine seeks lazily after\n    # the consumer is positioned at the partition's earliest offset, so any\n    # user-supplied 'auto.offset.reset' value that doesn't already mean\n    # \"start at the beginning\" is silently rewritten on the Rust side.\n    # Surface that rewrite explicitly so somebody who picked 'latest' on","sourceCodeStart":257,"sourceCodeEnd":293,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L257-L293","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Pass a Unix timestamp in milliseconds: start_from_timestamp_ms=int(time.time() * 1000) - 3600_000 for one hour ago.","Use None (or omit) when you don't want timestamp-based seeking — do not use -1 as a sentinel.","Double-check units: Kafka wants milliseconds, not seconds."],"exampleFix":"# before\nimport time\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", start_from_timestamp_ms=-1)\n# after\nimport time\nt = pw.io.kafka.read(\n    rdkafka_settings,\n    topic=\"t\",\n    start_from_timestamp_ms=int(time.time() * 1000) - 3_600_000,\n)","handlingStrategy":"validation","validationCode":"if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:\n    raise SystemExit(\"start_from_timestamp_ms must be a non-negative Unix ms timestamp\")\n\n# helper: build safely from a datetime\nimport time\ndef ms_ago(seconds: int) -> int:\n    return max(0, int(time.time() * 1000) - seconds * 1000)","typeGuard":"def valid_start_ts(v) -> bool:\n    return v is None or (isinstance(v, int) and v >= 0)","tryCatchPattern":null,"preventionTips":["Always multiply seconds by 1000 — Kafka wants milliseconds.","Use None (not -1) to disable timestamp seeking.","Note Pathway forces auto.offset.reset to 'earliest' when this is set, so the seek can fall back to the partition start."],"tags":["kafka","validation","timestamp","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}