{"record":{"id":"19da477dbc261b9c","repo":"apache/beam","slug":"current-source-r-and-a-source-created-using-the-range-of-the","errorCode":null,"errorMessage":"Current source %r and a source created using the range of the primary source %r determined by performing dynamic work rebalancing at fraction %r produced different values. Expected these sources to produce the same list of values.","messagePattern":"Current source %r and a source created using the range of the primary source %r determined by performing dynamic work rebalancing at fraction %r produced different values\\. Expected these sources to produce the same list of values\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/source_test_utils.py","lineNumber":388,"sourceCode":"    split_successful,\n    primary_range,\n    residual_range,\n    split_fraction):\n\n  assert primary_range\n  primary_items = read_from_source(source, *primary_range)\n\n  if not split_successful:\n    # For unsuccessful splits, residual_range should be None.\n    assert not residual_range\n\n  residual_items = (\n      read_from_source(source, *residual_range) if split_successful else [])\n\n  total_items = primary_items + residual_items\n\n  if current_items != primary_items:\n    raise ValueError(\n        'Current source %r and a source created using the '\n        'range of the primary source %r determined '\n        'by performing dynamic work rebalancing at fraction '\n        '%r produced different values. Expected '\n        'these sources to produce the same list of values.' %\n        (source, _range_to_str(*primary_range), split_fraction))\n\n  if expected_items != total_items:\n    raise ValueError(\n        'Items obtained by reading the source %r for primary '\n        'and residual ranges %s and %s did not produce the '\n        'expected list of values.' %\n        (source, _range_to_str(*primary_range), _range_to_str(*residual_range)))\n\n  result = (len(primary_items), len(residual_items) if split_successful else -1)\n  return result\n\n","sourceCodeStart":370,"sourceCodeEnd":406,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/source_test_utils.py#L370-L406","documentation":"Thrown by _verify_single_split_fraction_result when the items read from the original source differ from the items read from the primary range produced by split_at_fraction. Dynamic work rebalancing must be value-preserving: the primary range alone must reproduce exactly the items that would have been read up to the split point.","triggerScenarios":"Calling assert_split_at_fraction_behavior/binary (or the concurrent variant) on a custom source whose RangeTracker produces a primary range whose re-read items don't match the original iteration — typically a broken position<->item mapping in the RangeTracker or a non-deterministic read.","commonSituations":"Custom BoundedSource whose RangeTracker.start/stop positions don't align with record boundaries; source whose read depends on external mutable state (nondeterministic data); incorrect consumption of split positions after implementing try_split.","solutions":["Fix the source's RangeTracker so positions map deterministically to record boundaries (position must be the position AFTER returning an element).","Verify read_from_source(source, start, split_stop) yields exactly the first k elements the unsplit read produced.","Make the source deterministic — no reliance on unordered external data during the test.","Use assert_split_at_fraction_binary with a simple deterministic source to isolate whether the RangeTracker or the data is at fault."],"exampleFix":"// before (RangeTracker returns position of record start)\ndef position_at_fraction(self, fraction):\n  return int(fraction * self.stop_position)  # misaligned\n// after\ndef position_at_fraction(self, fraction):\n  return self.start_position + int(fraction * (self.stop_position - self.start_position))  # clamp to record boundaries in try_split","handlingStrategy":"validation","validationCode":"expected_prefix = items[:k]\nprimary_only = read_from_source(source, start, split_stop)\nassert primary_only == expected_prefix, 'primary range is not value-preserving'","typeGuard":null,"tryCatchPattern":"try:\n    assert_split_at_fraction_behavior(source, outcome, fraction, k)\nexcept ValueError as e:\n    if 'produced different values' in str(e):\n        logging.error('Primary range inconsistent with unsplit read: %s', e)","preventionTips":["Keep RangeTracker positions aligned to record boundaries (position = position after each returned element)","Make test sources fully deterministic","Split-return tests should use the same reader implementation the production pipeline uses"],"tags":["python","apache-beam","iobase","dynamic-work-rebalancing","consistency"],"backgroundTag":"internal-invariant-violation","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"}