{"record":{"id":"6f496235c43d0bd6","repo":"apache/beam","slug":"error-retrieving-metadata-v","errorCode":null,"errorMessage":"error retrieving metadata: %v","messagePattern":"error retrieving metadata: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/natsio/read.go","lineNumber":216,"sourceCode":"\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\n\t\t\tcount++\n\t\t}\n\n\t\tif err := msgs.Error(); err != nil {\n\t\t\treturn sdf.StopProcessing(), fmt.Errorf(\"error in message batch: %v\", err)\n\t\t}\n","sourceCodeStart":198,"sourceCodeEnd":234,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/natsio/read.go#L198-L234","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify messages actually flow through the JetStream stream (nats stream info <stream> shows message counts) rather than being published core-NATS only.","Upgrade github.com/nats-io/nats.go to a version matching your NATS server (JetStream APIs changed across versions).","Remove or fix any proxy/middleware that touches NATS reply subjects between client and server.","If caused by a specific poison message, identify its subject/sequence from logs and republish or purge it from the stream.","As a last resort, catch and skip unparseable messages in a forked version of this DoFn instead of failing the bundle."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// ensure traffic actually flows through JetStream before reading\njs, _ := jetstream.New(nc)\ninfo, err := js.StreamInfo(ctx, stream)\nif err != nil || info.State.Msgs == 0 {\n    return fmt.Errorf(\"stream %s missing or empty: %w\", stream, err)\n}","typeGuard":null,"tryCatchPattern":"metadata, err := msg.Metadata()\nif err != nil {\n    log.Warnf(\"skipping non-JetStream message on %s: %v\", msg.Subject(), err)\n    continue // or fail loudly if every message should have metadata\n}","preventionTips":["Never publish directly to a subject covered by the stream's pull consumer outside JetStream publish paths","Keep nats.go client version aligned with your NATS server version","Avoid proxies/middleware that rewrite NATS reply subjects","Test with `nats pub` vs `nats publish --js` to confirm JetStream ingress"],"tags":["nats","jetstream","metadata","protocol"],"backgroundTag":"unexpected-response-shape","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"}