{"record":{"id":"7c2ca9268d184f39","repo":"apache/beam","slug":"watermark-must-be-monotonally-increasing-is-at-v-got-v","errorCode":null,"errorMessage":"watermark must be monotonally increasing, is at %v, got %v","messagePattern":"watermark must be monotonally increasing, is at (.+?), got (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/testing/teststream/teststream.go","lineNumber":77,"sourceCode":"// SetEndpoint sets a URL for a TestStreamService that will emit events instead of having them\n// defined manually. Currently does not support authentication, so the TestStreamService should\n// be accessed in a trusted context.\nfunc (c *Config) setEndpoint(url string) {\n\tc.endpoint.Url = url\n}\n\n// createPayload converts the Config object into a TestStreamPayload to be sent to the runner.\nfunc (c *Config) createPayload() *pipepb.TestStreamPayload {\n\t// c0 is always the first coder in the pipeline, and inserting the TestStream as the first\n\t// element in the pipeline guarantees that the c0 coder corresponds to the type it outputs.\n\treturn &pipepb.TestStreamPayload{CoderId: \"c0\", Events: c.events, Endpoint: c.endpoint}\n}\n\n// AdvanceWatermark adds an event to the Config Events struct advancing the watermark for the PCollection\n// to the given timestamp. Timestamp is in milliseconds\nfunc (c *Config) AdvanceWatermark(timestamp int64) error {\n\tif c.watermark >= timestamp {\n\t\treturn fmt.Errorf(\"watermark must be monotonally increasing, is at %v, got %v\", c.watermark, timestamp)\n\t}\n\twatermarkAdvance := &pipepb.TestStreamPayload_Event_AdvanceWatermark{NewWatermark: timestamp}\n\twatermarkEvent := &pipepb.TestStreamPayload_Event_WatermarkEvent{WatermarkEvent: watermarkAdvance}\n\tc.events = append(c.events, &pipepb.TestStreamPayload_Event{Event: watermarkEvent})\n\tc.watermark = timestamp\n\treturn nil\n}\n\n// AdvanceWatermarkToInfinity advances the watermark to the maximum timestamp.\nfunc (c *Config) AdvanceWatermarkToInfinity() error {\n\treturn c.AdvanceWatermark(mtime.MaxTimestamp.Milliseconds())\n}\n\n// AdvanceProcessingTime adds an event advancing the processing time by a given duration.\n// This advancement is applied to all of the PCollections output by the TestStream.\nfunc (c *Config) AdvanceProcessingTime(duration int64) {\n\tprocessingAdvance := &pipepb.TestStreamPayload_Event_AdvanceProcessingTime{AdvanceDuration: duration}\n\tprocessingEvent := &pipepb.TestStreamPayload_Event_ProcessingTimeEvent{ProcessingTimeEvent: processingAdvance}","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/testing/teststream/teststream.go#L59-L95","documentation":"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.","triggerScenarios":"Calling Config.AdvanceWatermark(ts) with ts <= the previously advanced watermark (including the initial watermark or after AdvanceWatermarkToInfinity).","commonSituations":"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.","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"],"exampleFix":"// before\ncfg.AdvanceWatermark(100)\ncfg.AdvanceWatermark(100) // error\n// after\ncfg.AdvanceWatermark(100)\ncfg.AdvanceWatermark(200)","handlingStrategy":"validation","validationCode":"if ts <= lastWatermark { return fmt.Errorf(\"watermark %d must be > %d\", ts, lastWatermark) } // before cfg.AdvanceWatermark(ts)","typeGuard":null,"tryCatchPattern":"if err := cfg.AdvanceWatermark(ts); err != nil {\n    // monotonicity violated: fix event ordering\n    t.Fatalf(\"bad watermark advance: %v\", err)\n}","preventionTips":["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"],"tags":["go","testing","teststream","watermark"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}