[Bug][Go]: Metrics incremented in Setup methods are not recalled.

What happened?

Because the ParDo Setup context is uncached metrics initialized in Setup are lost, which is unexpected. Work done in setup, while logically outside of a bundle, will be under the context of the first bundle to execute that transform.

So there needs to be a way to transfer/extract the metrics from the Setup context so they are recorded back to the runner.

Issue Priority

Priority: 3 (minor)

Issue Components

@lostluck

Hi @lostluck,

I am interested in this issue and have been exploring Apache Beam metrics before working on this issue regarding unrecalled metric increments in Setup methods. As part of my learning, I wrote a simple Beam Go SDK program that converts words to uppercase and ran it on a local Flink development cluster. The dev environment looks like Apache Beam Go SDK v2.62.0, apache/beam_flink1.19_job_server, and flink:1.19.0-java11.

I observed that no custom metrics are reported in the Flink runner—only internal Beam metrics appear in the Accumulators tab. This behavior seems related to issue #32895.

Any insights on this?

For reference, here is the simple program I used:

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"reflect"
	"strings"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/flink"
)

// Define metrics.
var (
	processedCount   = metrics.NewCounter("example", "processed_count")
	invalidCount     = metrics.NewCounter("example", "invalid_count")
	sizeDistribution = metrics.NewDistribution("example", "record_size")
)

// UppercaseFn is a DoFn that processes words and updates metrics.
type UppercaseFn struct{}

func (fn *UppercaseFn) ProcessElement(word string, emit func(string)) {
	processedCount.Inc(context.Background(), 1)
	sizeDistribution.Update(context.Background(), int64(len(word)))

	if word == "" {
		invalidCount.Inc(context.Background(), 1)
		return
	}

	emit(strings.ToUpper(word))
}

// PrintFn is a DoFn that prints each element.
type PrintFn struct{}

func (fn *PrintFn) ProcessElement(word string) {
	fmt.Println(word)
}

func init() {
	beam.RegisterType(reflect.TypeOf((*UppercaseFn)(nil)).Elem())
	beam.RegisterType(reflect.TypeOf((*PrintFn)(nil)).Elem())
}

func main() {
	flag.Parse()
	beam.Init()

	pipeline := beam.NewPipeline()
	scope := pipeline.Root()

	inputData := []string{"apple", "banana", "", "cherry"}
	words := beam.CreateList(scope, inputData)

	uppercaseWords := beam.ParDo(scope, &UppercaseFn{}, words)
	beam.ParDo0(scope, &PrintFn{}, uppercaseWords)

	result, err := flink.Execute(context.Background(), pipeline)
	if err != nil {
		log.Fatalf("Failed to execute pipeline: %v", err)
	}

	metrics := result.Metrics().AllMetrics()
	for _, counter := range metrics.Counters() {
		fmt.Printf("Counter %s: %d\n", counter.Name(), counter.Committed)
	}

	for _, distribution := range metrics.Distributions() {
		fmt.Printf("Distribution %s: min=%d, max=%d, sum=%d, mean=%d, count=%d\n",
			distribution.Name(),
			distribution.Committed.Min,
			distribution.Committed.Max,
			distribution.Committed.Sum,
			distribution.Committed.Sum/distribution.Committed.Count,
			distribution.Committed.Count)
	}
}

Looking forward to your thoughts! 🙏

Sorry for missing this earlier @mohamedawnallah .

If Flink doesn't support exporting user metrics, then it's not something the Go SDK can fix by itself.

IIRC this issue is strictly for when a DoFn has metrics, and increments them in in the Once Per DoFn Instance Setup() lifecycle method. The provided code would not verify this issue by itself since it has no Setup methods on its dofns.

The default Go SDK runner, Prism, should be able to validate this issue though, if the base code reveals it. The changes would be in the exec package, ultimately.

.free-issue

.take-issue

.free-issue

Hi, I'd like to work on this. I'll submit a focused PR shortly.

After tracing the current metrics lifecycle, this needs a broader design around carrying one-time setup metrics into the first bundle store than I can safely complete as a small focused patch right now, so I'm stepping back to leave it available. Sorry for the noise.

Hi! I'd like to work on this issue. Please let me know if I can be assigned. Thanks!

Hi! I'd like to fix Go metrics incremented in Setup methods not being recalled. Plan: reproduce, fix the metric lifecycle, add a test. May I be assigned?

@LeonxLJX

Thank you for your interest @LeonxLJX ! You can self assign most issues by following the instructions at the contribution guide: https://github.com/apache/beam/blob/master/CONTRIBUTING.md#share-your-intent

Hi, I'd like to pick this up and investigate the metrics loss in Setup methods.

Hi, I'd like to pick this up. I'll look into the metrics initialization in the Setup method for the Go SDK.

Hi, I'd like to pick this up and work on a fix for capturing metrics from the Setup() lifecycle method.

Hi, I'd like to pick this up and work on ensuring metrics incremented in Setup() are correctly recorded.

@kragent66-glitch

Thank you for your interest @kragent66-glitch ! You can self assign most issues by following the instructions at the contribution guide: https://github.com/apache/beam/blob/master/CONTRIBUTING.md#share-your-intent

(Yes I'm teasing the obvious AI bot).

@lostluck

Hi @lostluck, after investigating, I found that this issue is related to the Setup() method not receiving the bundle context when it doesn't accept a context.Context argument. I've written a test case that confirms if Setup(ctx context.Context) is used, the metric is correctly captured. If the user uses Setup() without context and passes context.Background() to the metric API, the metric is correctly (but perhaps unexpectedly) dropped because context.Background() has no counter set. I'll provide a PR with a test case to clarify this convention.

Hi, I'd like to work on this issue as a first contribution . Could you please assign it to me?

@singhmrinal02

@singhmrinal02 There's an open PR attempting to address the issue, I would recommend checking out another open issue. As stated above, you can self-assign issues by following https://github.com/apache/beam/blob/master/CONTRIBUTING.md#share-your-intent

Merge setup metrics tests into one file and document the behavior und…

Add test case verifying metric capture in Setup() lifecycle method (#…

Navigation Menu

[Bug][Go]: Metrics incremented in Setup methods are not recalled. #27038

What happened?

Issue Priority

Issue Components

4 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[Bug][Go]: Metrics incremented in Setup methods are not recalled. #27038

Description

What happened?

Issue Priority

Issue Components

Activity

moawnallah commented on Feb 7, 2025

lostluck commented on Feb 20, 2025

albertjaison18 commented on Jul 13, 2026

mgalore commented on Jul 22, 2026

mgalore commented on Jul 22, 2026

4 remaining items

nightcityblade commented on Aug 13, 2026

nightcityblade commented on Aug 13, 2026

LeonxLJX commented on Aug 27, 2026

LeonxLJX commented on Sep 1, 2026

lostluck commented on Sep 2, 2026

kragent66-glitch commented on Sep 2, 2026

kragent66-glitch commented on Sep 2, 2026

kragent66-glitch commented on Sep 2, 2026

kragent66-glitch commented on Sep 2, 2026

lostluck commented on Sep 2, 2026

kragent66-glitch commented on Sep 2, 2026

singhmrinal02 commented on Sep 16, 2026

jrmccluskey commented on Sep 17, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions