apache/beam · error · BeamAssertException

Failed assert: [] == %r

Error message

Failed assert: [] == %r

What it means

The is_empty() matcher's _empty function converts the actual PCollection contents to a list and raises BeamAssertException if the list is non-empty, reporting that the empty list [] did not equal the actual elements. It asserts a pipeline stage produced zero output.

Solutions

1. Fix the upstream filter/transform so no elements pass through to the assertion.
2. If output is legitimately non-empty, replace is_empty() with equal_to(expected_elements).
3. Inspect the actual elements in the message to identify which records escaped filtering.

Example fix

// before
assert_that(pcoll, is_empty())  # pcoll has ['x']
// after
assert_that(pcoll, equal_to([]))  # or fix the filter upstream

# assert upstream filter predicate actually excludes all sample records
assert all(not keep(record) for record in sample_input), 'filter leaks records'

from apache_beam.testing.util import BeamAssertException
try:
    assert_that(pcoll, is_empty())
except BeamAssertException as e:
    logging.error('Expected empty output, got elements: %s', e)
    raise

When it happens

Trigger: assert_that(pcoll, is_empty()) where the pipeline emitted one or more elements — e.g. a filter intended to remove everything let some records through.

Common situations: Testing that invalid/filtered records are dropped; regression tests asserting no side outputs; overly permissive filter predicates; unexpected data in test input.

Related errors

• Failed assert: element
• Failed assert: %r == %r
• Failed assert: Received element
• Failed assert: unmatched elements
• Failed assert: window

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

Appendix: source

Thrown at sdks/python/apache_beam/testing/util.py:298

  Args:
    expected: A list of elements or hamcrest matchers to be used to match
      the elements of a single PCollection.
  """
  def _matches(actual):
    from hamcrest.core import assert_that as hamcrest_assert
    from hamcrest.library.collection import contains_inanyorder
    expected_list = list(expected)

    hamcrest_assert(actual, contains_inanyorder(*expected_list))

  return _matches


def is_empty():
  def _empty(actual):
    actual = list(actual)
    if actual:
      raise BeamAssertException('Failed assert: [] == %r' % actual)

  return _empty


def is_not_empty():
  """
  This is test method which makes sure that the pcol is not empty and it has
  some data in it.
  :return:
  """
  def _not_empty(actual):
    actual = list(actual)
    if not actual:
      raise BeamAssertException('Failed assert: pcol is empty')

  return _not_empty

View on GitHub (pinned to 12126d8942)