apache/beam · error · ValueError

SyntheticSource currently only supports delay distributions…

Error message

SyntheticSource currently only supports delay distributions of type 'const'. Received %s.

What it means

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.

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.

Example fix

// before
input_spec = {'delayDistribution': {'type': 'uniform', 'min': 1, 'max': 10}}
// after
input_spec = {'delayDistribution': {'type': 'const', 'const': 5}}

Defensive patterns

Strategy: validation

Validate before calling

dd = input_spec.get('delayDistribution')
if dd is not None and dd.get('type') != 'const':
    raise ValueError('delayDistribution.type must be const')

Type guard

def has_valid_delay_distribution(spec: dict) -> bool:
    dd = spec.get('delayDistribution')
    return dd is None or dd.get('type') == 'const'

Try / catch

try:
    source = SyntheticStep(input_spec, ...)
except ValueError as e:
    if 'delay distributions of type' in str(e):
        input_spec['delayDistribution'] = {'type': 'const', 'const': input_spec['delayDistribution'].get('const', 0)}
        source = SyntheticStep(input_spec, ...)
    else:
        raise

Prevention

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

When it happens

Trigger: Passing input_spec = {..., 'delayDistribution': {'type': 'uniform', ...}} or {'type': 'zipf', ...} to SyntheticSource / synthetic pipeline step construction.

Common situations: 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'.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/41d0515e9298d8d5. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/testing/synthetic_pipeline.py:384

        input_spec['forceNumInitialBundles']
        if 'forceNumInitialBundles' in input_spec else 0)
    if self._initial_splitting == 'zipf':
      self._initial_splitting_distribution_parameter = (
          input_spec['bundleSizeDistribution']['param'])
      if self._initial_splitting_distribution_parameter < 1:
        raise ValueError(
            'Parameter for a Zipf distribution must be larger than 1. '
            'Received %r.',
            self._initial_splitting_distribution_parameter)
    else:
      self._initial_splitting_distribution_parameter = 0
    self._dynamic_splitting = (
        'none' if (
            'splitPointFrequencyRecords' in input_spec and
            input_spec['splitPointFrequencyRecords'] == 0) else 'perfect')
    if 'delayDistribution' in input_spec:
      if input_spec['delayDistribution']['type'] != 'const':
        raise ValueError(
            'SyntheticSource currently only supports delay '
            'distributions of type \'const\'. Received %s.',
            input_spec['delayDistribution']['type'])
      self._sleep_per_input_record_sec = (
          float(input_spec['delayDistribution']['const']) / 1000)
      if (self._sleep_per_input_record_sec and
          self._sleep_per_input_record_sec < 1e-3):
        raise ValueError(
            'Sleep time per input record must be at least 1e-3.'
            ' Received: %r',
            self._sleep_per_input_record_sec)
    else:
      self._sleep_per_input_record_sec = 0

    # algorithm of the generator
    self.gen_algo = input_spec.get('algorithm', None)
    if self.gen_algo not in (None, 'builtin', 'lcg'):
      raise ValueError(

View on GitHub (pinned to 12126d8942)