apache/beam · error

err

Error message

err

What it means

natsio's endEstimator.Estimate panics when getEndSeqNo fails to fetch the stream's latest sequence number from JetStream. The estimate is used internally for sizing a read range, and there is no error channel in the estimator interface, so failure escalates to a panic. This indicates a JetStream API/connectivity problem, not a data issue.

Solutions

1. Verify the stream name passed to natsio matches an existing JetStream stream (nats stream info <name>)
2. Confirm JetStream is enabled on the server and reachable with the given credentials
3. Check that the NATS connection context/auth config is valid from the execution environment
4. Wrap the estimator call with recover() or pre-check the stream existence with js.StreamInfo yourself

Example fix

// before
end := estimator.Estimate() // panics if stream missing
// after
if _, err := js.StreamInfo(ctx, streamName); err != nil {
    log.Fatalf("stream %q not available: %v", streamName, err)
}
end := estimator.Estimate()

if _, err := js.StreamInfo(ctx, streamName); err != nil {
    return fmt.Errorf("stream %q unavailable: %w", streamName, err)
}

defer func() { if r := recover(); r != nil { log.Fatalf("natsio estimate: %v", r) } }()

When it happens

Trigger: Calling Estimate (during pipeline setup/estimation) when the NATS server is unreachable, the stream name does not exist (js.Stream returns ErrStreamNotFound), credentials are rejected, or the JetStream request times out.

Common situations: Typo in stream name; NATS server without JetStream enabled; network/auth problems between the worker and NATS; stream deleted after connection was established.

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 publishing message
• error fetching messages
• natsio.Read: invalid option
• unable to dial sdk worker pool
• unsupported time policy

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

Appendix: source

Thrown at sdks/go/pkg/beam/io/natsio/end_estimator.go:44

type endEstimator struct {
	js      jetstream.JetStream
	stream  string
	subject string
}

func newEndEstimator(js jetstream.JetStream, stream string, subject string) *endEstimator {
	return &endEstimator{
		js:      js,
		stream:  stream,
		subject: subject,
	}
}

func (e *endEstimator) Estimate() int64 {
	ctx := context.Background()
	end, err := e.getEndSeqNo(ctx)
	if err != nil {
		panic(err)
	}
	return end
}

func (e *endEstimator) getEndSeqNo(ctx context.Context) (int64, error) {
	str, err := e.js.Stream(ctx, e.stream)
	if err != nil {
		return -1, fmt.Errorf("error getting stream: %v", err)
	}

	msg, err := str.GetLastMsgForSubject(ctx, e.subject)
	if err != nil {
		if isMessageNotFound(err) {
			return 1, nil
		}

		return -1, fmt.Errorf("error getting last message: %v", err)
	}

View on GitHub (pinned to 12126d8942)