feat(inputs.nats_consumer): add durable consumer support for JetStream by aleksclark · Pull Request #18720 · influxdata/telegraf

aleksclark · 2026-04-15T14:55:00Z

Summary

Add durable consumer support to the NATS JetStream input plugin, allowing consumers to survive Telegraf restarts without message loss or replay.

The existing JetStream support uses ephemeral push consumers. When Telegraf restarts, the consumer is lost and messages may be replayed from the beginning of the stream or missed entirely. Durable consumers persist on the NATS server and resume from where they left off.

New Configuration Options

Option	Type	Default	Description
`jetstream_durable_name`	string	`""`	Durable consumer name — survives restarts
`jetstream_deliver_policy`	string	`"all"`	Where to start: all, last, new, by_start_sequence, by_start_time
`jetstream_start_sequence`	uint64	`0`	Starting sequence for by_start_sequence policy
`jetstream_start_time`	string	`""`	RFC3339 start time for by_start_time policy
`jetstream_ack_wait`	duration	`30s`	Server-side ack timeout before redelivery
`jetstream_max_deliver`	int	`-1`	Max redelivery attempts (-1 = unlimited)
`jetstream_filter_subjects`	[]string	`[]`	Filter subjects within a stream
`jetstream_consumer_name`	string	`""`	Explicit consumer name for pre-created consumers

Behavior Changes

When durable_name is set, uses Drain() instead of Unsubscribe() on shutdown to preserve the consumer
Correctly uses js.Subscribe() when no queue group is set, instead of always using QueueSubscribe()

Tests

6 new integration tests covering: durable restart resilience, deliver policies (new, by_start_sequence), max deliver limits, filter subjects, and durable + queue group combinations
All tests use testcontainers with the official nats Docker image

Files Changed

plugins/inputs/nats_consumer/nats_consumer.go — New config fields + subscription options
plugins/inputs/nats_consumer/nats_consumer_test.go — 6 new integration tests (580 lines)
plugins/inputs/nats_consumer/sample.conf — New config options documentation
plugins/inputs/nats_consumer/README.md — Updated documentation

Checklist

No AI generated code was used in this PR
AI generated code used in this PR follows the InfluxData Policy on AI-Generated Code Contributions

Related issues

resolves #18734

Add support for durable JetStream consumers that survive Telegraf restarts, preventing message loss or replay. New config options: - jetstream_durable_name: bind to a named durable consumer - jetstream_deliver_policy: control where in the stream to start (all, last, new, by_start_sequence, by_start_time) - jetstream_start_sequence / jetstream_start_time: for sequence/time-based delivery - jetstream_ack_wait: server-side ack timeout before redelivery - jetstream_max_deliver: max redelivery attempts per message - jetstream_filter_subjects: filter subjects within a stream - jetstream_consumer_name: explicit consumer name for pre-created consumers When durable_name is set, subscriptions use Drain() instead of Unsubscribe() on shutdown to preserve the consumer on the server. The plugin also now uses js.Subscribe() when no queue group is set, instead of always using QueueSubscribe(). 🐨 Generated with Crush Assisted-by: AWS Claude Opus 4.6 via Crush <crush@charm.land>

telegraf-tiger · 2026-04-15T14:55:07Z

Thanks so much for the pull request!
🤝 ✒️ Just a reminder that the CLA has not yet been signed, and we'll need it before merging. Please sign the CLA when you get a chance, then post a comment here saying !signed-cla

aleksclark · 2026-04-15T19:22:34Z

!signed-cla

🐘 Generated with Crush Assisted-by: AWS Claude Opus 4.6 via Crush <crush@charm.land>

telegraf-tiger · 2026-04-15T19:46:55Z

Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

☺️ This pull request doesn't significantly change the Telegraf binary size (less than 1%)

📦 Click here to get additional PR build artifacts

Artifact URLs

. DEB	. RPM	. TAR . GZ	. ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz

srebhan · 2026-04-17T09:43:52Z

@aleksclark can you please restore the PR description template, especially the AI section, as we cannot review your PR otherwise!

aleksclark · 2026-04-17T11:11:53Z

@srebhan sure thing, sorry about that!

srebhan

Thanks for your contribution @aleksclark! I do have some comments in the code.
One question came up on if jetstream_consumer_name and jetstream_filter_subjects are required for the durable stream feature or if those are separate things. If they are separate, please move them to dedicated PRs!

srebhan · 2026-04-22T11:50:36Z

+	JsAckWait              config.Duration `toml:"jetstream_ack_wait"`
+	JsMaxDeliver           int             `toml:"jetstream_max_deliver"`
+	JsFilterSubjects       []string        `toml:"jetstream_filter_subjects"`
+	JsConsumerName         string          `toml:"jetstream_consumer_name"`


Does it make sense to move this to an own structure?

srebhan · 2026-04-22T11:53:29Z

+	validPolicies := []string{"", "all", "last", "new", "by_start_sequence", "by_start_time"}
+	if !slices.Contains(validPolicies, n.JsDeliverPolicy) {
+		return fmt.Errorf("invalid jetstream_deliver_policy %q, must be one of: all, last, new, by_start_sequence, by_start_time", n.JsDeliverPolicy)
+	}


The following is much easier to read

Suggested change

validPolicies := []string{"", "all", "last", "new", "by_start_sequence", "by_start_time"}

if !slices.Contains(validPolicies, n.JsDeliverPolicy) {

return fmt.Errorf("invalid jetstream_deliver_policy %q, must be one of: all, last, new, by_start_sequence, by_start_time", n.JsDeliverPolicy)

}

switch n.JsDeliverPolicy {

case "":

n.JsDeliverPolicy = "all"

case "all", "last", "new", "by_start_sequence", "by_start_time":

// Do nothing, those are valid

default:

return fmt.Errorf("invalid 'jetstream_deliver_policy' %q", n.JsDeliverPolicy)

}

Same for the checks below.

We really should get rid of all other ways of checking this...

srebhan · 2026-04-22T11:54:06Z

+	}
+
+	if n.JsDeliverPolicy == "by_start_sequence" && n.JsStartSequence == 0 {
+		return errors.New("jetstream_start_sequence must be set when jetstream_deliver_policy is \"by_start_sequence\"")


I would do

Suggested change

return errors.New("jetstream_start_sequence must be set when jetstream_deliver_policy is \"by_start_sequence\"")

return errors.New("jetstream_start_sequence must be set when jetstream_deliver_policy is 'by_start_sequence'")

to avoid the escaping. Same below.

srebhan · 2026-04-22T11:56:20Z

+				n.Log.Errorf("Error draining JetStream subject %s in queue %s: %s",
+					sub.Subject, sub.Queue, err)


Suggested change

n.Log.Errorf("Error draining JetStream subject %s in queue %s: %s",

sub.Subject, sub.Queue, err)

n.Log.Errorf("Error draining JetStream subject %s in queue %s: %s", sub.Subject, sub.Queue, err)

srebhan · 2026-04-22T11:56:32Z

+				n.Log.Errorf("Error unsubscribing from subject %s in queue %s: %s",
+					sub.Subject, sub.Queue, err)


Suggested change

n.Log.Errorf("Error unsubscribing from subject %s in queue %s: %s",

sub.Subject, sub.Queue, err)

n.Log.Errorf("Error unsubscribing from subject %s in queue %s: %s", sub.Subject, sub.Queue, err)

srebhan · 2026-04-22T12:05:18Z

+	}
+}
+
+func TestJetStreamDeliverPolicyNew(t *testing.T) {


Same as above!

The question is: Can't we turn this into a table-based test function to avoid the large amount of code duplication?

srebhan · 2026-04-22T12:07:24Z

+  ## jetstream durable consumer name
+  ## When set, the plugin creates or binds to a durable consumer that
+  ## survives restarts. Without this, an ephemeral consumer is created
+  ## and messages may be replayed or missed on restart.


Please be brief here

Suggested change

## jetstream durable consumer name

## When set, the plugin creates or binds to a durable consumer that

## survives restarts. Without this, an ephemeral consumer is created

## and messages may be replayed or missed on restart.

## When set, the plugin creates or binds to a durable consumer surviving restarts

Same below. Please remove the variable naming header line as this does not add any information!

srebhan · 2026-04-22T12:09:02Z

+  ## jetstream start sequence
+  ## Used with jetstream_deliver_policy = "by_start_sequence".
+  ## The consumer will start receiving messages from this stream sequence number.
+  # jetstream_start_sequence = 0


How would a user know what to put in here? Is this shown somewhere in NATS?

srebhan · 2026-04-22T12:10:13Z

+  ## Maximum number of delivery attempts for a single message. Set to -1
+  ## for unlimited redelivery attempts.
+  # jetstream_max_deliver = -1


Suggested change

## Maximum number of delivery attempts for a single message. Set to -1

## for unlimited redelivery attempts.

# jetstream_max_deliver = -1

## Maximum number of delivery attempts for a single message (default: unlimited)

# jetstream_max_deliver = 0

srebhan · 2026-04-22T12:11:00Z

+  ## jetstream filter subjects
+  ## Filter specific subjects within a stream (JetStream 2.10+).
+  ## This allows consuming only a subset of subjects from a stream.
+  # jetstream_filter_subjects = []
+
+  ## jetstream consumer name
+  ## Explicit consumer name for named consumers (different from durable_name).
+  ## Used for pre-created consumers or when you want a specific consumer identity.
+  # jetstream_consumer_name = ""


It seems like these two are unrelated to the durable stream feature, aren't they?

telegraf-tiger · 2026-05-11T18:09:41Z

Hello! I am closing this issue due to inactivity. I hope you were able to resolve your problem, if not please try posting this question in our Community Slack or Community Forums or provide additional details in this issue and reqeust that it be re-opened. Thank you!

telegraf-tiger Bot added feat Improvement on an existing feature such as adding a new setting/mode to an existing plugin plugin/input 1. Request for new input plugins 2. Issues/PRs that are related to input plugins labels Apr 15, 2026

fix(inputs.nats_consumer): address golangci-lint errors

fce9bd8

🐘 Generated with Crush Assisted-by: AWS Claude Opus 4.6 via Crush <crush@charm.land>

srebhan self-assigned this Apr 17, 2026

srebhan added the waiting for response waiting for response from contributor label Apr 17, 2026

aleksclark mentioned this pull request Apr 17, 2026

feat(inputs.nats_consumer): add durable consumer support for JetStream #18734

Open

telegraf-tiger Bot removed the waiting for response waiting for response from contributor label Apr 17, 2026

srebhan reviewed Apr 22, 2026

View reviewed changes

srebhan added the waiting for response waiting for response from contributor label Apr 27, 2026

telegraf-tiger Bot closed this May 11, 2026

-	validPolicies := []string{"", "all", "last", "new", "by_start_sequence", "by_start_time"}
-	if !slices.Contains(validPolicies, n.JsDeliverPolicy) {
-		return fmt.Errorf("invalid jetstream_deliver_policy %q, must be one of: all, last, new, by_start_sequence, by_start_time", n.JsDeliverPolicy)
-	}
+	switch n.JsDeliverPolicy {
+		case "":
+			n.JsDeliverPolicy = "all"
+		case "all", "last", "new", "by_start_sequence", "by_start_time":
+			// Do nothing, those are valid
+		default:
+			return fmt.Errorf("invalid 'jetstream_deliver_policy' %q", n.JsDeliverPolicy)
+	}

	return errors.New("jetstream_start_sequence must be set when jetstream_deliver_policy is \"by_start_sequence\"")
	return errors.New("jetstream_start_sequence must be set when jetstream_deliver_policy is 'by_start_sequence'")

		n.Log.Errorf("Error draining JetStream subject %s in queue %s: %s",
		sub.Subject, sub.Queue, err)

		n.Log.Errorf("Error unsubscribing from subject %s in queue %s: %s",
		sub.Subject, sub.Queue, err)

Conversation

aleksclark commented Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

New Configuration Options

Behavior Changes

Tests

Files Changed

Checklist

Related issues

Uh oh!

telegraf-tiger Bot commented Apr 15, 2026

Uh oh!

aleksclark commented Apr 15, 2026

Uh oh!

telegraf-tiger Bot commented Apr 15, 2026

Artifact URLs

Uh oh!

srebhan commented Apr 17, 2026

Uh oh!

aleksclark commented Apr 17, 2026

Uh oh!

srebhan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

telegraf-tiger Bot commented May 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

aleksclark commented Apr 15, 2026 •

edited

Loading