watermark must be monotonally increasing, is at

Error message

watermark must be monotonally increasing, is at %v, got %v

What it means

TestStream configs replay a fixed event list; watermarks must advance strictly monotonically because Beam's watermark model only moves forward. AdvanceWatermark rejects a timestamp that is not strictly greater than the config's current watermark.

Solutions

Ensure each AdvanceWatermark timestamp is strictly greater than all previous ones
Reorder events so watermark advances are ascending
Track the last advanced timestamp in test setup code before appending events
Start from a lower initial watermark and advance incrementally

Example fix

// before
cfg.AdvanceWatermark(100)
cfg.AdvanceWatermark(100) // error
// after
cfg.AdvanceWatermark(100)
cfg.AdvanceWatermark(200)

Defensive patterns

Strategy: validation

Validate before calling

if ts <= lastWatermark { return fmt.Errorf("watermark %d must be > %d", ts, lastWatermark) } // before cfg.AdvanceWatermark(ts)

Try / catch

if err := cfg.AdvanceWatermark(ts); err != nil {
    // monotonicity violated: fix event ordering
    t.Fatalf("bad watermark advance: %v", err)
}

Prevention

Track the last advanced watermark in test helper code
Sort events by timestamp before appending
Never advance after AdvanceWatermarkToInfinity
Build TestStream events through a helper that enforces ordering

When it happens

Trigger: Calling Config.AdvanceWatermark(ts) with ts <= the previously advanced watermark (including the initial watermark or after AdvanceWatermarkToInfinity).

Common situations: Copy-pasting watermark events with the same timestamp, sorting events incorrectly when building a TestStream programmatically, or accidentally calling AdvanceWatermarkToInfinity and then trying to advance further.

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

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

Appendix: source

Thrown at sdks/go/pkg/beam/testing/teststream/teststream.go:77

// SetEndpoint sets a URL for a TestStreamService that will emit events instead of having them
// defined manually. Currently does not support authentication, so the TestStreamService should
// be accessed in a trusted context.
func (c *Config) setEndpoint(url string) {
	c.endpoint.Url = url
}

// createPayload converts the Config object into a TestStreamPayload to be sent to the runner.
func (c *Config) createPayload() *pipepb.TestStreamPayload {
	// c0 is always the first coder in the pipeline, and inserting the TestStream as the first
	// element in the pipeline guarantees that the c0 coder corresponds to the type it outputs.
	return &pipepb.TestStreamPayload{CoderId: "c0", Events: c.events, Endpoint: c.endpoint}
}

// AdvanceWatermark adds an event to the Config Events struct advancing the watermark for the PCollection
// to the given timestamp. Timestamp is in milliseconds
func (c *Config) AdvanceWatermark(timestamp int64) error {
	if c.watermark >= timestamp {
		return fmt.Errorf("watermark must be monotonally increasing, is at %v, got %v", c.watermark, timestamp)
	}
	watermarkAdvance := &pipepb.TestStreamPayload_Event_AdvanceWatermark{NewWatermark: timestamp}
	watermarkEvent := &pipepb.TestStreamPayload_Event_WatermarkEvent{WatermarkEvent: watermarkAdvance}
	c.events = append(c.events, &pipepb.TestStreamPayload_Event{Event: watermarkEvent})
	c.watermark = timestamp
	return nil
}

// AdvanceWatermarkToInfinity advances the watermark to the maximum timestamp.
func (c *Config) AdvanceWatermarkToInfinity() error {
	return c.AdvanceWatermark(mtime.MaxTimestamp.Milliseconds())
}

// AdvanceProcessingTime adds an event advancing the processing time by a given duration.
// This advancement is applied to all of the PCollections output by the TestStream.
func (c *Config) AdvanceProcessingTime(duration int64) {
	processingAdvance := &pipepb.TestStreamPayload_Event_AdvanceProcessingTime{AdvanceDuration: duration}
	processingEvent := &pipepb.TestStreamPayload_Event_ProcessingTimeEvent{ProcessingTimeEvent: processingAdvance}

View on GitHub (pinned to 12126d8942)