{"record":{"id":"cb6c02c98805523a","repo":"pathwaycom/pathway","slug":"synchronization-group-can-only-be-set-once","errorCode":null,"errorMessage":"synchronization_group can only be set once","messagePattern":"synchronization_group can only be set once","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/internals/datasource.py","lineNumber":28,"sourceCode":"import pandas as pd\n\nfrom pathway.internals import api\nfrom pathway.internals.schema import Schema, schema_from_pandas\n\n\n@dataclass(frozen=True)\nclass DataSourceOptions:\n    commit_duration_ms: int | None = None\n    unsafe_trusted_ids: bool | None = False\n    unique_name: str | None = None\n    synchronization_group: api.ConnectorGroupDescriptor | None = None\n    max_backlog_size: int | None = None\n\n    def set_synchronization_group(self, group: api.ConnectorGroupDescriptor | None):\n        if self.synchronization_group is None:\n            object.__setattr__(self, \"synchronization_group\", group)\n        else:\n            raise ValueError(\"synchronization_group can only be set once\")\n\n\n@dataclass(frozen=True, kw_only=True)\nclass DataSource(ABC):\n    schema: type[Schema]\n    data_source_options: DataSourceOptions = DataSourceOptions()\n\n    @property\n    def connector_properties(self) -> api.ConnectorProperties:\n        columns: list[api.ColumnProperties] = []\n        for column in self.schema.columns().values():\n            columns.append(\n                api.ColumnProperties(\n                    dtype=column.dtype.to_engine(),\n                    append_only=self.is_append_only(),\n                )\n            )\n","sourceCodeStart":10,"sourceCodeEnd":46,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/internals/datasource.py#L10-L46","documentation":"DataSourceOptions is a frozen dataclass whose set_synchronization_group assigns a connector synchronization group exactly once. Calling it a second time (or when synchronization_group was already set at construction) raises, because two conflicting synchronization-group assignments would make commit-ordering semantics ambiguous.","triggerScenarios":"Calling source.options.set_synchronization_group(g1) and later set_synchronization_group(g2); applying a synchronization group to a source that already got one from the read API (e.g. pw.io.csv.read(..., synchronization_group=...) followed by another set).","commonSituations":"Framework code that wraps Pathway connectors and sets a default sync group, then user code sets its own; refactored pipelines calling a helper that sets the group twice.","solutions":["Set the synchronization group once — pass it directly to the read function: pw.io.csv.read(path, synchronization_group=my_group).","Guard before setting: if source.data_source_options.synchronization_group is None: ...set... .","If a different group is genuinely needed, construct a new source instead of mutating the existing one."],"exampleFix":"# before\nopts.set_synchronization_group(group_a)\n...\nopts.set_synchronization_group(group_b)  # raises\n\n# after\n# decide the group up front and set it only once\npw.io.csv.read(path, synchronization_group=group_b)","handlingStrategy":"validation","validationCode":"assert source.data_source_options.synchronization_group is None, 'synchronization group already assigned; construct a new source instead'","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Set the group once via the read() kwarg instead of imperative mutation.","Wrap set_synchronization_group with an is-None guard in framework code."],"tags":["pathway","connectors","synchronization","api-misuse"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}