{"record":{"id":"41d0515e9298d8d5","repo":"apache/beam","slug":"syntheticsource-currently-only-supports-delay-distributions","errorCode":null,"errorMessage":"SyntheticSource currently only supports delay distributions of type 'const'. Received %s.","messagePattern":"SyntheticSource currently only supports delay distributions of type 'const'\\. Received (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/testing/synthetic_pipeline.py","lineNumber":384,"sourceCode":"        input_spec['forceNumInitialBundles']\n        if 'forceNumInitialBundles' in input_spec else 0)\n    if self._initial_splitting == 'zipf':\n      self._initial_splitting_distribution_parameter = (\n          input_spec['bundleSizeDistribution']['param'])\n      if self._initial_splitting_distribution_parameter < 1:\n        raise ValueError(\n            'Parameter for a Zipf distribution must be larger than 1. '\n            'Received %r.',\n            self._initial_splitting_distribution_parameter)\n    else:\n      self._initial_splitting_distribution_parameter = 0\n    self._dynamic_splitting = (\n        'none' if (\n            'splitPointFrequencyRecords' in input_spec and\n            input_spec['splitPointFrequencyRecords'] == 0) else 'perfect')\n    if 'delayDistribution' in input_spec:\n      if input_spec['delayDistribution']['type'] != 'const':\n        raise ValueError(\n            'SyntheticSource currently only supports delay '\n            'distributions of type \\'const\\'. Received %s.',\n            input_spec['delayDistribution']['type'])\n      self._sleep_per_input_record_sec = (\n          float(input_spec['delayDistribution']['const']) / 1000)\n      if (self._sleep_per_input_record_sec and\n          self._sleep_per_input_record_sec < 1e-3):\n        raise ValueError(\n            'Sleep time per input record must be at least 1e-3.'\n            ' Received: %r',\n            self._sleep_per_input_record_sec)\n    else:\n      self._sleep_per_input_record_sec = 0\n\n    # algorithm of the generator\n    self.gen_algo = input_spec.get('algorithm', None)\n    if self.gen_algo not in (None, 'builtin', 'lcg'):\n      raise ValueError(","sourceCodeStart":366,"sourceCodeEnd":402,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/testing/synthetic_pipeline.py#L366-L402","documentation":"SyntheticSource can simulate per-record processing delay, but its implementation only knows how to sleep a constant amount of time per record. If input_spec contains a 'delayDistribution' whose 'type' is anything other than 'const' (e.g. 'zipf' or 'uniform'), __init__ raises ValueError because that distribution type is unsupported.","triggerScenarios":"Passing input_spec = {..., 'delayDistribution': {'type': 'uniform', ...}} or {'type': 'zipf', ...} to SyntheticSource / synthetic pipeline step construction.","commonSituations":"Copy-pasting a bundleSizeDistribution config (which supports zipf) into the delayDistribution field; assuming delay distributions are as flexible as size distributions; typos like 'constant' instead of 'const'.","solutions":["Change delayDistribution type to 'const' and provide the delay in milliseconds under the 'const' key.","Remove the delayDistribution key entirely if no per-record delay is needed (defaults to 0 sleep).","If variable delay is genuinely required, subclass or patch SyntheticSource to implement the desired distribution."],"exampleFix":"// before\ninput_spec = {'delayDistribution': {'type': 'uniform', 'min': 1, 'max': 10}}\n// after\ninput_spec = {'delayDistribution': {'type': 'const', 'const': 5}}","handlingStrategy":"validation","validationCode":"dd = input_spec.get('delayDistribution')\nif dd is not None and dd.get('type') != 'const':\n    raise ValueError('delayDistribution.type must be const')","typeGuard":"def has_valid_delay_distribution(spec: dict) -> bool:\n    dd = spec.get('delayDistribution')\n    return dd is None or dd.get('type') == 'const'","tryCatchPattern":"try:\n    source = SyntheticStep(input_spec, ...)\nexcept ValueError as e:\n    if 'delay distributions of type' in str(e):\n        input_spec['delayDistribution'] = {'type': 'const', 'const': input_spec['delayDistribution'].get('const', 0)}\n        source = SyntheticStep(input_spec, ...)\n    else:\n        raise","preventionTips":["Only ever emit {'type': 'const', 'const': <ms>} for delayDistribution in config generators","Do not copy bundleSizeDistribution shapes into delayDistribution","Pin the apache_beam version and check SyntheticSource docs for supported options"],"tags":["python","apache-beam","valueerror","synthetic-pipeline","unsupported-feature"],"backgroundTag":"unsupported-enum-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}