apache/beam · error

error in message batch

Error message

error in message batch: %v

What it means

Returned when msgs.Error() reports a batch-level error after iterating a JetStream fetch result. Fetch can partially succeed (some messages delivered) while a request error is pending — e.g. consumer server-side reset failure, permission error on a subject, or expiration errors. The DoFn stops processing to let Beam retry.

Solutions

1. Inspect the wrapped error (%v) to identify whether it is a permissions, no-responders, or consumer-reset error and address accordingly.
2. Verify account/user permissions allow both subscribe on the subject and consumer creation on the stream (nats account info / server config).
3. Confirm the stream still exists and is healthy (nats stream info); recreate or restore it if purged/deleted.
4. Increase OrderedConsumerConfig.MaxResetAttempts if failovers are frequent in your cluster.
5. Rely on Beam bundle retry: because TryClaim sequences progress, reprocessing resumes from the last claimed sequence.

// pre-check permissions and stream presence before the pipeline
si, err := js.StreamInfo(ctx, stream)
if err != nil { return err }
// confirm the account may create consumers
if si.Config.Subjects == nil { return errors.New("stream has no subjects") }

if err := msgs.Error(); err != nil {
    var permErr *nats.Error
    if errors.As(err, &permErr) && permErr.Error() == nats.ErrAuthorization.Error() {
        return sdf.StopProcessing(), fmt.Errorf("permission denied during fetch batch: %w", err)
    }
    return sdf.StopProcessing(), fmt.Errorf("error in message batch: %v", err)
}

When it happens

Trigger: After consuming msgs.Messages(), msgs.Error() returns non-nil at read.go:231. Happens when the fetch encountered per-request errors such as jetstream.ErrNoResponses style failures, consumer reset exhaustion on an OrderedConsumer, or a permissions (AuthorizationViolation) error on the delivery subject mid-batch.

Common situations: Stream purged or deleted while the pipeline is running; NATS account permissions changed removing read access to the subject; server failover causing the ordered consumer to exceed MaxResetAttempts (5); disk-jetstream file store failures on the server.

Related errors

• error creating consumer
• err
• error creating growable tracker
• error creating JetStream context
• error fetching messages

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

Appendix: source

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

			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)

			count++
		}

		if err := msgs.Error(); err != nil {
			return sdf.StopProcessing(), fmt.Errorf("error in message batch: %v", err)
		}

		if count == 0 {
			fn.updateWatermarkManually(we)
			return sdf.ResumeProcessingIn(resumeDelay), nil
		}
	}
}

func (fn *readFn) createRTracker(rest offsetrange.Restriction) (sdf.RTracker, error) {
	if rest.End < math.MaxInt64 {
		return offsetrange.NewTracker(rest), nil
	}

	estimator := newEndEstimator(fn.js, fn.Stream, fn.Subject)
	rt, err := offsetrange.NewGrowableTracker(rest, estimator)
	if err != nil {
		return nil, fmt.Errorf("error creating growable tracker: %v", err)

View on GitHub (pinned to 12126d8942)