Error when running redpanda connect in multiple pods with NATS Jetstream as input.

[BenBasilTomy]

I am running a redpanda connect with NATS Jetstream as input and getting an error:
"Failed to connect to nats_jetstream: cannot create a queue subscription for a consumer without a deliver group."
But as per redpanda connect documentation there is no config field for NATS Jetstream input as deliver group. How to solve this? Below is my input configuration:

input:
  label: orders_publisher
  nats_jetstream:
    urls: [nats://nats:4222]
    subject: awsloadtest
    stream: orders
    queue: testqueue
    durable: awsloadtest 
    ack_wait: 30s
    max_ack_pending: 1000
    deliver: all

[mihaitodor]

Hey @BenBasilTomy 👋 Would you mind also sharing how to spin up a NATS server locally (maybe via Docker / Docker Compose or inside a Kind Kubernetes cluster) and how to set it up to reproduce this issue?

[BenBasilTomy]

Hey @mihaitodor
I had asked a similar question in the NATS community and learned that it was more of a NATS issue than a Benthos one. The problem was caused by using both the queue and durable fields simultaneously. Removing the durable field while running Benthos with NATS as input in multiple pods solved the issue.

The NATS community mentioned that the error message indicated I was trying to subscribe to a consumer using a queue group ('queue': testqueue in my configuration), which is only needed when subscribing to a push consumer. However, the consumer I was trying to use ('durable': awsloadtest in my config) was configured as a pull consumer. They suggested checking the Benthos documentation for how to configure it to get data from a pull consumer.

I checked the documentation and could not find any configuration differentiating push and pull consumers.

[mihaitodor]

Nice catch! Sounds like something worth documenting. Please feel free to submit a PR if you wish. The docs are generated from here:

connect/internal/impl/nats/input_jetstream.go

Lines 60 to 97 in 968786d

	` + connectionNameDescription() + authDescription()).
	Fields(connectionHeadFields()...).
	Field(service.NewStringField("queue").
	Description("An optional queue group to consume as.").
	Optional()).
	Field(service.NewStringField("subject").
	Description("A subject to consume from. Supports wildcards for consuming multiple subjects. Either a subject or stream must be specified.").
	Optional().
	Example("foo.bar.baz").Example("foo.*.baz").Example("foo.bar.*").Example("foo.>")).
	Field(service.NewStringField("durable").
	Description("Preserve the state of your consumer under a durable name.").
	Optional()).
	Field(service.NewStringField("stream").
	Description("A stream to consume from. Either a subject or stream must be specified.").
	Optional()).
	Field(service.NewBoolField("bind").
	Description("Indicates that the subscription should use an existing consumer.").
	Optional()).
	Field(service.NewStringAnnotatedEnumField("deliver", map[string]string{
	"all": "Deliver all available messages.",
	"last": "Deliver starting with the last published messages.",
	"last_per_subject": "Deliver starting with the last published message per subject.",
	"new": "Deliver starting from now, not taking into account any previous messages.",
	}).
	Description("Determines which messages to deliver when consuming without a durable subscriber.").
	Default("all")).
	Field(service.NewStringField("ack_wait").
	Description("The maximum amount of time NATS server should wait for an ack from consumer.").
	Advanced().
	Default("30s").
	Example("100ms").
	Example("5m")).
	Field(service.NewIntField("max_ack_pending").
	Description("The maximum number of outstanding acks to be allowed before consuming is halted.").
	Advanced().
	Default(1024)).
	Fields(connectionTailFields()...).
	Field(inputTracingDocs())

(you'll have to run make docs to parse the code and generate the docs text files which also have to be committed in the PR). Additionally, it may be worth adding a linting rule so Benthos will complain about this invalid config on startup. Here's an example:

connect/internal/impl/snowflake/output_snowflake_put.go

Line 236 in 968786d

LintRule(`root = match {

[BenBasilTomy]

Thanks for the suggestion! I'll be happy to submit a PR to document this. I'll also look into adding a linting rule so Benthos will complain about this invalid config on startup.

[BenBasilTomy]

Hi @mihaitodor I’ve implemented the changes based on our conversation. Could you please review the PR?

https://github.com/redpanda-data/connect/pull/2760/files

[mihaitodor]

Thank you @BenBasilTomy! I left a small comment there.