apache/beam · error

error creating consumer

Error message

error creating consumer: %v

What it means

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.

Solutions

1. Verify the stream exists in the connected account: nats stream info <stream>; create it if missing.
2. Grant the credentials permission to JetStream API consumer creation (CONSUMER.CREATE on the stream) and subject read access.
3. Check that StartSeqNo (default 1) does not exceed the stream's current last sequence; adjust the StartSeqNo option.
4. Confirm connectivity/timeout: increase context timeout or fix network so the API request to $JS.API.CONSUMER.CREATE completes.
5. If streams live on a different account or domain, pass the correct JetStream account/domain in the client configuration.

Example fix

// before
// pipeline starts before the stream exists
// after
// provision the stream first, e.g.:
js.CreateStream(ctx, jetstream.StreamConfig{Name: "my-stream", Subjects: []string{"orders.>"}})
// then run natsio.Read(s, uri, "my-stream", "orders.new")

// preflight: stream must exist and account must allow consumers
js, err := jetstream.New(nc)
if err != nil { return err }
if _, err := js.StreamInfo(ctx, stream); err != nil {
    return fmt.Errorf("stream %q not found: %w", stream, err)
}
if startSeqNo > int64(si.State.LastSeq) {
    return fmt.Errorf("start seq %d beyond stream last seq %d", startSeqNo, si.State.LastSeq)
}

cons, err := fn.js.OrderedConsumer(ctx, fn.Stream, cfg)
if err != nil {
    if errors.Is(err, jetstream.ErrStreamNotFound) {
        return nil, fmt.Errorf("stream %q does not exist; create it before reading", fn.Stream)
    }
    return nil, fmt.Errorf("error creating consumer: %v", err)
}

When it happens

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

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

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• error fetching messages
• error in message batch
• err
• error creating growable tracker
• error creating JetStream context

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

Appendix: source

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

	}

	return rt, nil
}

func (fn *readFn) createConsumer(
	ctx context.Context,
	startSeqNo int64,
) (jetstream.Consumer, error) {
	cfg := jetstream.OrderedConsumerConfig{
		FilterSubjects:   []string{fn.Subject},
		DeliverPolicy:    jetstream.DeliverByStartSequencePolicy,
		OptStartSeq:      uint64(startSeqNo),
		MaxResetAttempts: 5,
	}

	cons, err := fn.js.OrderedConsumer(ctx, fn.Stream, cfg)
	if err != nil {
		return nil, fmt.Errorf("error creating consumer: %v", err)
	}

	return cons, nil
}

func createConsumerMessage(msg jetstream.Msg, publishingTime time.Time) ConsumerMessage {
	return ConsumerMessage{
		Subject:        msg.Subject(),
		PublishingTime: publishingTime,
		ID:             msg.Headers().Get(nats.MsgIdHdr),
		Headers:        msg.Headers(),
		Data:           msg.Data(),
	}
}

func (fn *readFn) updateWatermarkManually(we *watermarkEstimator) {
	t := time.Now().Add(-1 * assumedLag)
	et := fn.timestampFn(t)

View on GitHub (pinned to 12126d8942)