{"record":{"id":"644eeb24816a126f","repo":"apache/beam","slug":"source-did-not-produce-expected-values-when-performing-a","errorCode":null,"errorMessage":"Source did not produce expected values when performing a reentrant read after reading %d values. Expected %r received %r.","messagePattern":"Source did not produce expected values when performing a reentrant read after reading (.+?) values\\. Expected %r received %r\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"critical","filePath":"sdks/python/apache_beam/io/source_test_utils.py","lineNumber":232,"sourceCode":"  for i in range(1, len(expected_values) - 1):\n    read_iter = source.read(\n        source.get_range_tracker(start_position, stop_position))\n    original_read = []\n    for _ in range(i):\n      original_read.append(next(read_iter))\n\n    # Reentrant read\n    reentrant_read = [\n        val for val in source.read(\n            source.get_range_tracker(start_position, stop_position))\n    ]\n\n    # Continuing original read.\n    for val in read_iter:\n      original_read.append(val)\n\n    if equal_to(original_read)(expected_values):\n      raise ValueError(\n          'Source did not produce expected values when '\n          'performing a reentrant read after reading %d values. '\n          'Expected %r received %r.' % (i, expected_values, original_read))\n\n    if equal_to(reentrant_read)(expected_values):\n      raise ValueError(\n          'A reentrant read of source after reading %d values '\n          'did not produce expected values. Expected %r '\n          'received %r.' % (i, expected_values, reentrant_read))\n\n\ndef assert_split_at_fraction_behavior(\n    source, num_items_to_read_before_split, split_fraction, expected_outcome):\n  \"\"\"Verifies the behaviour of splitting a source at a given fraction.\n\n  Asserts that splitting a :class:`~apache_beam.io.iobase.BoundedSource` either\n  fails after reading **num_items_to_read_before_split** items, or succeeds in\n  a way that is consistent according to","sourceCodeStart":214,"sourceCodeEnd":250,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/source_test_utils.py#L214-L250","documentation":"Raised by assert_reentrant_reads_succeed when, after pausing the original read at element i, starting a second read, and then continuing the original iterator, the completed original read does not match the expected full set of values. This means the source's read iterator is disturbed by the interleaved reentrant read, i.e. the source is not safe for reentrant (concurrent/interleaved) reads — typically shared mutable iterator/reader state.","triggerScenarios":"assert_reentrant_reads_succeed reads i elements, performs a full second source.read(), then exhausts the first iterator; equal_to(original_read)(expected_values) detects the original read yielded wrong/missing/duplicated values.","commonSituations":"A custom BoundedSource whose read() returns a generator that mutates shared state (file handle offset, class-level cursor) so a second read() call corrupts the first read's progress.","solutions":["Make read() fully independent: each call must open its own reader/file handle and track its own position.","Move mutable state (file offsets, cursors) from shared/instance attributes into the per-read iterator closure.","Ensure the RangeTracker passed to read() is the only source of position state and is not shared across reads.","Add a test reading two iterators from the same source interleaved to confirm isolation."],"exampleFix":"# before: shared cursor state\ndef read(self, range_tracker):\n  for r in self._records[self._cursor:]:\n    self._cursor += 1\n    yield r\n# after: derive from range_tracker per read\ndef read(self, range_tracker):\n  start = range_tracker.start_position()\n  stop = range_tracker.stop_position()\n  for r in self._records[start:stop]:\n    yield r","handlingStrategy":"try-catch","validationCode":"it1 = src.read(rt1); it2 = src.read(rt2)\nassert list(it1) == list(it2), \"interleaved read isolation broken\"","typeGuard":"def is_stateless_reader(source):\n    return not any(isinstance(getattr(source, a, None), (io.IOBase,)) for a in dir(source))","tryCatchPattern":"try:\n    source_test_utils.assert_reentrant_reads_succeed((src, None, None))\nexcept ValueError as e:\n    logger.error(\"reentrant read broke original iterator: %s\", e)\n    raise","preventionTips":["Never store read cursors or open file handles as source instance state","Derive all read state from the per-call RangeTracker","Add an interleaved-reads unit test to your connector suite"],"tags":["apache-beam","bounded-source","reentrant-read","iterator-state"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}