{"record":{"id":"b0908b73c49145d5","repo":"apache/beam","slug":"error-fetching-messages-v","errorCode":null,"errorMessage":"error fetching messages: %v","messagePattern":"error fetching messages: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/natsio/read.go","lineNumber":209,"sourceCode":"}\n\nfunc (fn *readFn) ProcessElement(\n\tctx context.Context,\n\twe *watermarkEstimator,\n\trt *sdf.LockRTracker,\n\t_ []byte,\n\temit func(beam.EventTime, ConsumerMessage),\n) (sdf.ProcessContinuation, error) {\n\tstartSeqNo := rt.GetRestriction().(offsetrange.Restriction).Start\n\tcons, err := fn.createConsumer(ctx, startSeqNo)\n\tif err != nil {\n\t\treturn sdf.StopProcessing(), err\n\t}\n\n\tfor {\n\t\tmsgs, err := cons.Fetch(fn.FetchSize, jetstream.FetchMaxWait(fetchTimeout))\n\t\tif err != nil {\n\t\t\treturn nil, fmt.Errorf(\"error fetching messages: %v\", err)\n\t\t}\n\n\t\tcount := 0\n\t\tfor msg := range msgs.Messages() {\n\t\t\tmetadata, err := msg.Metadata()\n\t\t\tif err != nil {\n\t\t\t\treturn sdf.StopProcessing(), fmt.Errorf(\"error retrieving metadata: %v\", err)\n\t\t\t}\n\n\t\t\tseqNo := int64(metadata.Sequence.Stream)\n\t\t\tif !rt.TryClaim(seqNo) {\n\t\t\t\treturn sdf.StopProcessing(), nil\n\t\t\t}\n\n\t\t\tet := fn.timestampFn(metadata.Timestamp)\n\t\t\tconsMsg := createConsumerMessage(msg, metadata.Timestamp)\n\t\t\temit(et, consMsg)\n","sourceCodeStart":191,"sourceCodeEnd":227,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/natsio/read.go#L191-L227","documentation":"This error is returned by the NATS JetStream source DoFn (readFn.ProcessElement) when cons.Fetch fails. Fetch performs a batch pull request against a JetStream ordered consumer; any transport-level failure (connection dropped, context expired, timeout with no messages, consumer reset exhausted) is wrapped here. It aborts the bundle with a hard error rather than resuming later.","triggerScenarios":"cons.Fetch(fn.FetchSize, jetstream.FetchMaxWait(fetchTimeout)) at read.go:207 returns err. Occurs when the NATS connection is closed/unhealthy, the consumer was reset more than MaxResetAttempts (5) times, the fetch exceeds FetchMaxWait (3s) with an underlying transport error, or the server rejects the pull request (e.g. stream deleted mid-read).","commonSituations":"NATS server restart or network partition during a long-running streaming pipeline; stream or subject deleted or renamed while the pipeline runs; credentials expiring mid-run; worker losing connection to a remote NATS cluster in a distributed environment.","solutions":["Check NATS server connectivity and health from the worker (nats server ping / connection options); ensure the server URL in natsio.Read is reachable from all runners.","Verify the stream name passed to natsio.Read exists and the subject filter matches messages (nats stream info).","Increase reliability: run NATS with a cluster, and configure connection retry options (nats.MaxReconnects, nats.ReconnectWait) in the underlying connection setup.","For transient errors, rerun the pipeline; Beam will retry failed bundles, and the source claims sequence numbers via the restriction tracker so work resumes safely.","Check credentials/authorization validity for the duration of the job if using CredsFile."],"exampleFix":"// before\nopts := []nats.Option{}\n// after\nopts := []nats.Option{\n  nats.MaxReconnects(-1),\n  nats.ReconnectWait(2 * time.Second),\n  nats.Timeout(10 * time.Second),\n}\n// pass these when creating the connection used by natsio.Read so transient\n// outages reconnect instead of surfacing as fetch errors","handlingStrategy":"retry","validationCode":"// before building the pipeline\nconn, err := nats.Connect(uri, nats.MaxReconnects(-1), nats.ReconnectWait(2*time.Second))\nif err != nil {\n    return fmt.Errorf(\"NATS unreachable at %s: %w\", uri, err)\n}\nconn.Close()","typeGuard":null,"tryCatchPattern":"if _, err := cons.Fetch(size, jetstream.FetchMaxWait(3*time.Second)); err != nil {\n    if nats.IsReconnecting(err) || errors.Is(err, nats.ErrNoResponders) {\n        return sdf.ResumeProcessingIn(5 * time.Second), nil // transient: retry later\n    }\n    return sdf.StopProcessing(), fmt.Errorf(\"error fetching messages: %v\", err)\n}","preventionTips":["Configure reconnect options on the NATS connection used by the source","Alert on NATS server health; run a clustered server for production pipelines","Validate stream/subject names with `nats stream info` before launching the job","Use credentials with long-lived validity or rotate them outside pipeline runs"],"tags":["nats","jetstream","network","fetch","streaming"],"backgroundTag":"network-request-failed","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"}