apache/beam · error

error retrieving metadata

Error message

error retrieving metadata: %v

What it means

Returned when msg.Metadata() fails for a message received from the JetStream fetch. Metadata decodes the reply subject containing delivery/stream sequence numbers and timestamps; failure means the message's reply subject is not a valid JetStream acknowledgment token. The DoFn stops processing the bundle.

Solutions

1. Verify messages actually flow through the JetStream stream (nats stream info <stream> shows message counts) rather than being published core-NATS only.
2. Upgrade github.com/nats-io/nats.go to a version matching your NATS server (JetStream APIs changed across versions).
3. Remove or fix any proxy/middleware that touches NATS reply subjects between client and server.
4. If caused by a specific poison message, identify its subject/sequence from logs and republish or purge it from the stream.
5. As a last resort, catch and skip unparseable messages in a forked version of this DoFn instead of failing the bundle.

// ensure traffic actually flows through JetStream before reading
js, _ := jetstream.New(nc)
info, err := js.StreamInfo(ctx, stream)
if err != nil || info.State.Msgs == 0 {
    return fmt.Errorf("stream %s missing or empty: %w", stream, err)
}

metadata, err := msg.Metadata()
if err != nil {
    log.Warnf("skipping non-JetStream message on %s: %v", msg.Subject(), err)
    continue // or fail loudly if every message should have metadata
}

When it happens

Trigger: msg.Metadata() inside the msgs.Messages() loop returns err — practically only when a non-JetStream message is delivered to the consumer, e.g. a message published directly to a subject consumed by the pull consumer rather than flowing through the stream, or a malformed/empty reply subject from a misbehaving server or proxy.

Common situations: A NATS proxy or leaf-node setup rewriting reply subjects; messages injected by tooling directly on the subject; version mismatch between nats.go client and server producing incompatible reply tokens; custom interception middleware corrupting the inbox reply.

Related errors

• err
• error creating consumer
• 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/6f496235c43d0bd6. Report an issue: GitHub.

Appendix: source

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

	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)

			count++
		}

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

View on GitHub (pinned to 12126d8942)