apache/beam · error

error fetching messages

Error message

error fetching messages: %v

What it means

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.

Solutions

1. 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.
2. Verify the stream name passed to natsio.Read exists and the subject filter matches messages (nats stream info).
3. Increase reliability: run NATS with a cluster, and configure connection retry options (nats.MaxReconnects, nats.ReconnectWait) in the underlying connection setup.
4. 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.
5. Check credentials/authorization validity for the duration of the job if using CredsFile.

Example fix

// before
opts := []nats.Option{}
// after
opts := []nats.Option{
  nats.MaxReconnects(-1),
  nats.ReconnectWait(2 * time.Second),
  nats.Timeout(10 * time.Second),
}
// pass these when creating the connection used by natsio.Read so transient
// outages reconnect instead of surfacing as fetch errors

// before building the pipeline
conn, err := nats.Connect(uri, nats.MaxReconnects(-1), nats.ReconnectWait(2*time.Second))
if err != nil {
    return fmt.Errorf("NATS unreachable at %s: %w", uri, err)
}
conn.Close()

if _, err := cons.Fetch(size, jetstream.FetchMaxWait(3*time.Second)); err != nil {
    if nats.IsReconnecting(err) || errors.Is(err, nats.ErrNoResponders) {
        return sdf.ResumeProcessingIn(5 * time.Second), nil // transient: retry later
    }
    return sdf.StopProcessing(), fmt.Errorf("error fetching messages: %v", err)
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• err
• error creating consumer
• error publishing message
• chunk send failed
• chunk send failed

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

Appendix: source

Thrown at sdks/go/pkg/beam/io/natsio/read.go:209

}

func (fn *readFn) ProcessElement(
	ctx context.Context,
	we *watermarkEstimator,
	rt *sdf.LockRTracker,
	_ []byte,
	emit func(beam.EventTime, ConsumerMessage),
) (sdf.ProcessContinuation, error) {
	startSeqNo := rt.GetRestriction().(offsetrange.Restriction).Start
	cons, err := fn.createConsumer(ctx, startSeqNo)
	if err != nil {
		return sdf.StopProcessing(), err
	}

	for {
		msgs, err := cons.Fetch(fn.FetchSize, jetstream.FetchMaxWait(fetchTimeout))
		if err != nil {
			return nil, fmt.Errorf("error fetching messages: %v", err)
		}

		count := 0
		for msg := range msgs.Messages() {
			metadata, err := msg.Metadata()
			if err != nil {
				return sdf.StopProcessing(), fmt.Errorf("error retrieving metadata: %v", err)
			}

			seqNo := int64(metadata.Sequence.Stream)
			if !rt.TryClaim(seqNo) {
				return sdf.StopProcessing(), nil
			}

			et := fn.timestampFn(metadata.Timestamp)
			consMsg := createConsumerMessage(msg, metadata.Timestamp)
			emit(et, consMsg)

View on GitHub (pinned to 12126d8942)