apache/beam · error

source failed processing timers

Error message

source failed processing timers

What it means

Same wrapping pattern as data errors but for timers: when a bundle event carries timers (len(e.Timers)>0), the `timer` callback processes them; any non-EOF error is wrapped as 'source failed processing timers'. It indicates the timer-processing path (DoFn ProcessElement with timer invocation, or timer decode) failed.

Solutions

1. Read the wrapped cause error to identify the failing timer callback
2. Verify timer family IDs and coders match between pipeline stages / after job update
3. Fix the user DoFn logic executed when the timer fires
4. Check for incompatibilities if the pipeline was updated while draining state

// timer callback errors surface as 'source failed processing timers'
if err := timerFn(ctx, key); err != nil {
    log.Errorf("timer processing failed: %v", err)
    return err
}

When it happens

Trigger: An event with e.Timers set arrives; r.Reset(e.Timers) then timer(&bcr, e.PtransformID, e.TimerFamilyID) returns a non-nil error other than io.EOF during timer decoding or DoFn timer invocation.

Common situations: Stateful/timer DoFns (e.g. windowed aggregation with timers) failing at runtime; timer family mismatch after pipeline update; user code erroring inside timer-handling ProcessElement; runner re-delivering timers with stale schema.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• ProcessElement uses a TimerProvider, but no Timer fields…
• unsupported WindowCoder Type
• bad coder kind
• bad CoGBK
• bad combinefn

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/datasource.go:155

			// Channel closed, so time to exit
			if !ok {
				return nil
			}
			if len(e.Data) > 0 {
				r.Reset(e.Data)
				err = data(&bcr, e.PtransformID)
			}
			if err != nil && err != io.EOF {
				return errors.Wrapf(err, "source failed processing data")
			}
			// Process any simultaneously sent timers.
			// If the data channel has split though
			if len(e.Timers) > 0 {
				r.Reset(e.Timers)
				err = timer(&bcr, e.PtransformID, e.TimerFamilyID)
			}
			if err != nil && err != io.EOF {
				return errors.Wrap(err, "source failed processing timers")
			}
			// io.EOF means the reader successfully drained.
			// We're ready for a new buffer.
		case <-ctx.Done():
			// now that it is done processing received data, we set it to false.
			n.consumingReceivedData.Store(false)
			return nil
		}
	}
}

// ByteCountReader is a passthrough reader that counts all the bytes read through it.
// It trusts the nested reader to return accurate byte information.
type byteCountReader struct {
	count  *int
	reader io.Reader
}

View on GitHub (pinned to 12126d8942)