IssueTracker

This will create a public issue which anybody can view and comment on.

Please provide as much information as possible. At least, this should include a description of your issue and steps to reproduce the problem. If possible please provide a summary of what steps or workarounds you have already tried, and any docs or articles you found (un)helpful.

Problem you have encountered:

Custom counters defined for a dataflow job are visible in the dataflow console as Function without handlling exception and not being exported to dashboard

When running DataFlow, cx observed that using exception handling results in Metrics recorded by the beam DoFn not being logged into the dashboard. Cx provided a minimal example below where the same exact DoFn is used under 3 conditions:

-no exception handling -timeout -use_subprocess.

The DoFn simply increments a counter and yields. Only the counter for the DoFn without exception handling is logged.

What you expected to happen:

Only metrics for DoFns without exception handling are logged.

Steps to reproduce:

run this minimal reproducible example in your env

--runner=DataflowRunne

from datetime import datetime

import apache_beam as beam
import structlog
from apache_beam.metrics import Metrics

logger = structlog.get_logger(__name__)


class CountAndYield(beam.DoFn):
    """A DoFn that counts and yields elements."""

    def __init__(self):
        beam.DoFn.__init__(self)
        self.counter = Metrics.counter(self.__class__, "Simple counter")

    # pylint: disable=arguments-differ
    def process(self, element):
        self.counter.inc()
        yield element


def apply_graph(pipeline: beam.Pipeline):
    elements = (
        pipeline
        | "Start with 3 examples" >> beam.Create([1, 2, 3])
        | "Function without Exception Handling" >> beam.ParDo(CountAndYield())
    )

    elements, _ = elements | "Function with Exception Handling (timeout)" >> beam.ParDo(
        CountAndYield()
    ).with_exception_handling(timeout=60 * 30)

    elements, _ = (
        elements
        | "Function with Exception Handling (use_subprocess)"
        >> beam.ParDo(CountAndYield()).with_exception_handling(use_subprocess=True)
    )


if __name__ == "__main__":
    current_time = datetime.now().strftime("%Y%m%d-%H%M")
    job_name = f"metrics-bug-minimal-example-{current_time}"

    # Utility function that outputs a set of pipeline options.
    p_options = get_pipeline_options(
        number_of_worker_harness_threads=2,
        job_name=job_name,
        machine_type="n1-standard-4",
    )

    with beam.Pipeline(options=p_options) as p:
        apply_graph(pipeline=p)

"""
class CountAndYield(beam.DoFn):
    """A DoFn that counts and yields elements."""

    def __init__(self):
        beam.DoFn.__init__(self)
        self.counter = Metrics.counter(self.__class__, "Simple counter")

    def process(self, element):
        try:
            self.counter.inc()
            yield element
        except Exception as e:
            logger.error(f"Error processing element {element}: {e}")
            self.counter.inc(-1)  # Decrement counter or handle as needed
            raise

Other information (workarounds you have tried, documentation consulted, etc):