{"record":{"id":"68e46a32cfcdc2a0","repo":"apache/beam","slug":"error-creating-consumer-v","errorCode":null,"errorMessage":"error creating consumer: %v","messagePattern":"error creating consumer: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/natsio/read.go","lineNumber":269,"sourceCode":"\t}\n\n\treturn rt, nil\n}\n\nfunc (fn *readFn) createConsumer(\n\tctx context.Context,\n\tstartSeqNo int64,\n) (jetstream.Consumer, error) {\n\tcfg := jetstream.OrderedConsumerConfig{\n\t\tFilterSubjects:   []string{fn.Subject},\n\t\tDeliverPolicy:    jetstream.DeliverByStartSequencePolicy,\n\t\tOptStartSeq:      uint64(startSeqNo),\n\t\tMaxResetAttempts: 5,\n\t}\n\n\tcons, err := fn.js.OrderedConsumer(ctx, fn.Stream, cfg)\n\tif err != nil {\n\t\treturn nil, fmt.Errorf(\"error creating consumer: %v\", err)\n\t}\n\n\treturn cons, nil\n}\n\nfunc createConsumerMessage(msg jetstream.Msg, publishingTime time.Time) ConsumerMessage {\n\treturn ConsumerMessage{\n\t\tSubject:        msg.Subject(),\n\t\tPublishingTime: publishingTime,\n\t\tID:             msg.Headers().Get(nats.MsgIdHdr),\n\t\tHeaders:        msg.Headers(),\n\t\tData:           msg.Data(),\n\t}\n}\n\nfunc (fn *readFn) updateWatermarkManually(we *watermarkEstimator) {\n\tt := time.Now().Add(-1 * assumedLag)\n\tet := fn.timestampFn(t)","sourceCodeStart":251,"sourceCodeEnd":287,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/natsio/read.go#L251-L287","documentation":"Returned by readFn.createConsumer (called from ProcessElement) when fn.js.OrderedConsumer(ctx, fn.Stream, cfg) fails to create or bind an ephemeral ordered consumer on the stream. Ordered consumers are server-created pull consumers with automatic reset; failure means the server rejected consumer creation for that stream/configuration.","triggerScenarios":"jetstream.OrderedConsumer returns err at read.go:267: the stream does not exist, the user lacks management/consumer permissions (no API access to CONSUMER.CREATE), the context is canceled/timed out, no responders for the API subject (stream gone / wrong account), or OptStartSeq (startSeqNo) exceeds the stream's last sequence.","commonSituations":"Typo'd stream name or wrong NATS account/jurisdiction (strip prefix) when streams live in a different account; restricted service-user permissions on shared NATS clusters; starting the pipeline against a server where the stream hasn't been provisioned yet; a start sequence number beyond existing data.","solutions":["Verify the stream exists in the connected account: nats stream info <stream>; create it if missing.","Grant the credentials permission to JetStream API consumer creation (CONSUMER.CREATE on the stream) and subject read access.","Check that StartSeqNo (default 1) does not exceed the stream's current last sequence; adjust the StartSeqNo option.","Confirm connectivity/timeout: increase context timeout or fix network so the API request to $JS.API.CONSUMER.CREATE completes.","If streams live on a different account or domain, pass the correct JetStream account/domain in the client configuration."],"exampleFix":"// before\n// pipeline starts before the stream exists\n// after\n// provision the stream first, e.g.:\njs.CreateStream(ctx, jetstream.StreamConfig{Name: \"my-stream\", Subjects: []string{\"orders.>\"}})\n// then run natsio.Read(s, uri, \"my-stream\", \"orders.new\")","handlingStrategy":"validation","validationCode":"// preflight: stream must exist and account must allow consumers\njs, err := jetstream.New(nc)\nif err != nil { return err }\nif _, err := js.StreamInfo(ctx, stream); err != nil {\n    return fmt.Errorf(\"stream %q not found: %w\", stream, err)\n}\nif startSeqNo > int64(si.State.LastSeq) {\n    return fmt.Errorf(\"start seq %d beyond stream last seq %d\", startSeqNo, si.State.LastSeq)\n}","typeGuard":null,"tryCatchPattern":"cons, err := fn.js.OrderedConsumer(ctx, fn.Stream, cfg)\nif err != nil {\n    if errors.Is(err, jetstream.ErrStreamNotFound) {\n        return nil, fmt.Errorf(\"stream %q does not exist; create it before reading\", fn.Stream)\n    }\n    return nil, fmt.Errorf(\"error creating consumer: %v\", err)\n}","preventionTips":["Provision streams with infrastructure-as-code before deploying the pipeline","Verify JetStream API permissions (CONSUMER.CREATE) for the service account","Check that StartSeqNo is within the stream's sequence range","Ensure the correct JetStream domain/account prefix is configured for multi-account setups","Smoke-test consumer creation with `nats con add` using the same credentials"],"tags":["nats","jetstream","consumer","permissions","streaming"],"backgroundTag":"resource-not-found","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"}