{"record":{"id":"dba74844e4758c27","repo":"pathwaycom/pathway","slug":"autocommit-duration-ms-must-be-positive-got-au","errorCode":null,"errorMessage":"'autocommit_duration_ms' must be positive; got {autocommit_duration_ms}. It is the maximum time between two commits and zero/negative values would prevent commits from happening.","messagePattern":"'autocommit_duration_ms' must be positive; got (.+?)\\. It is the maximum time between two commits and zero/negative values would prevent commits from happening\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":282,"sourceCode":"    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\n    # purpose doesn't see Pathway read from the beginning instead. The\n    # librdkafka aliases 'earliest', 'beginning' and 'smallest' all mean\n    # \"start at the beginning\" and therefore don't trigger the override.\n    _START_FROM_BEGINNING_ALIASES = {\"earliest\", \"beginning\", \"smallest\"}\n    user_offset_reset = rdkafka_settings.get(\"auto.offset.reset\")\n    if (\n        start_from_timestamp_ms is not None","sourceCodeStart":264,"sourceCodeEnd":300,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L264-L300","documentation":"autocommit_duration_ms in pw.io.kafka.read sets the maximum interval between consumer offset commits. Zero or negative values would mean commits never happen, so after a restart the consumer would re-read from its last persisted position (or the beginning) — Pathway rejects such values up front.","triggerScenarios":"pw.io.kafka.read(..., autocommit_duration_ms=0) or a negative value. None uses the connector default; only explicit non-positive integers raise.","commonSituations":"Setting autocommit_duration_ms=0 intending to 'disable autocommit' when the developer actually wants manual/None behavior; unit confusion (seconds vs ms) producing 0 after integer division.","solutions":["Use a positive millisecond value, e.g. autocommit_duration_ms=10000 for commits at least every 10 seconds.","If you don't need to tune committing, omit the parameter and let the default apply.","Never use 0 to mean 'off' — there is no supported way to disable commits entirely via this argument."],"exampleFix":"# before\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", autocommit_duration_ms=0)\n# after\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", autocommit_duration_ms=10_000)","handlingStrategy":"validation","validationCode":"if autocommit_duration_ms is not None and autocommit_duration_ms <= 0:\n    raise SystemExit(\"autocommit_duration_ms must be positive milliseconds or None\")\n\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", autocommit_duration_ms=autocommit_duration_ms)","typeGuard":"def valid_autocommit(v) -> bool:\n    return v is None or (isinstance(v, int) and v > 0)","tryCatchPattern":null,"preventionTips":["Use a sane default like 10_000 ms; do not set 0 hoping to disable commits.","Remember this value bounds at-least-once re-read volume after crashes — larger means fewer commits but more potential duplicates."],"tags":["kafka","validation","offsets","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}