Update dependencies & migrate to new jetstream module

[Jarema]

Signed-off-by: Tomasz Pietrek tomasz@synadia.com

[codecov]

Codecov Report

❌ Patch coverage is 63.33333% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.06%. Comparing base (310c39f) to head (5fe9982).
⚠️ Report is 5 commits behind head on main.

Files with missing lines	Patch %	Lines
server/core/connector.go	56.86%	16 Missing and 6 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #113      +/-   ##
==========================================
- Coverage   56.43%   56.06%   -0.38%     
==========================================
  Files          33       33              
  Lines        2151     2185      +34     
==========================================
+ Hits         1214     1225      +11     
- Misses        808      825      +17     
- Partials      129      135       +6     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Jarema]

@claude review this PR.

[synadia-claude-reviewer]

Claude finished @Jarema's task in 2m 3s —— View job

---

PR Review: Update dependencies & migrate to new jetstream module

• Gather context and understand the PR
• Read changed files
• Analyze changes for correctness, quality, and issues
• Post review feedback

---

This PR migrates from the legacy nats.JetStreamContext API to the new github.com/nats-io/nats.go/jetstream package, switching from push-based consumers to pull-based consumers. Overall the migration is solid, but there are a few issues worth addressing.

Bug: defer ackCancel() inside a message callback (connector.go:440)

cc, err := consumer.Consume(func(msg jetstream.Msg) {
    ...
    if ackSyncEnabled {
        ackCtx, ackCancel := context.WithTimeout(...)
        defer ackCancel()   // ← deferred to when this anonymous func returns
        if err := msg.DoubleAck(ackCtx); err != nil { ... }
    }
    ...
})

At first glance this looks fine (the defer runs when the callback returns), but defer in a loop or callback is a well-known Go gotcha. Here the Consume library likely invokes this callback sequentially per message, so the context is cancelled correctly on each invocation. However, the semantics are fragile: if the library ever invokes the callback asynchronously or reuses goroutines, the context could leak or be cancelled early. A direct ackCancel() call immediately after DoubleAck (replacing defer) would be safer and more explicit. Fix this →

Breaking change: QueueName silently dropped for JetStream connectors

Previously, conn.config.QueueName was passed to QueueSubscribe for push consumers. The new signature drops it entirely:

// Old
func (conn *BridgeConnector) subscribeToJetStream(subject string, queueName string) (*nats.Subscription, error)

// New  
func (conn *BridgeConnector) subscribeToJetStream(subject string) (jetstream.ConsumeContext, error)

Any existing deployments using QueueName in a JetStreamToKafka connector config will have that setting silently ignored. This should at minimum be documented (a warning log if QueueName is set), or alternatively the new pull consumer approach should implement shared consumers to preserve the semantic.

Breaking change: EnableFlowControl and HeartbeatInterval silently ignored

conf.JetStreamConfig still exposes EnableFlowControl bool and HeartbeatInterval int, but connectToJetStream() no longer applies them anywhere. These were push-consumer features and don't map 1:1 to pull consumers, but existing users with these settings in their config will get no indication they're being ignored. A deprecation warning in the logs would be appropriate.

Missing msg.Nak() on Kafka write failure (connector.go:431-433)

if err != nil {
    conn.stats.AddMessageIn(l)
    conn.bridge.Logger().Errorf("connector publish failure, ...")
    // message is not acked or nacked
}

When writing to Kafka fails, the message is left unacknowledged. It will eventually be redelivered after the ack timeout, which is the correct behavior, but it's inefficient. An explicit msg.Nak() or msg.NakWithDelay(...) would trigger immediate redelivery with backoff. This was a pre-existing issue but the migration to pull consumers makes it more relevant since delivery behavior is now fully controlled client-side.

Stats tracking asymmetry (connector.go:432)

conn.stats.AddMessageIn(l) is only called on the error path. On the success path, conn.stats.AddRequest(l, l, time.Since(start)) is called instead. This was present before and is likely intentional, but worth confirming the stats semantics are preserved correctly across the migration.

Test change: consumer pre-created with wrong policy (jetstream2kafka_test.go:463-469)

In TestJetStreamQueueDurableSubscriber, the test creates a consumer upfront:

_, err = tbs.JS.CreateOrUpdateConsumer(context.Background(), stream, jetstream.ConsumerConfig{
    Durable:       durable,
    AckPolicy:     jetstream.AckExplicitPolicy,
    DeliverPolicy: jetstream.DeliverAllPolicy,  // ← was DeliverByStartSequencePolicy
})

The old test used DeliverByStartSequencePolicy with OptStartSeq: 2 to verify the durable consumer resumed at the right position. The removal of StartAtSequence: 2 from the connector config and DeliverByStartSequencePolicy from the pre-created consumer simplifies the test but weakens it — it no longer tests that durable consumers resume from the correct offset. The same simplification is applied to the SASL variant.

Positive changes

• The ConsumeErrHandler addition is a good improvement — push consumers silently dropped errors.
• Auto-discovery of stream name via StreamNameBySubject is more ergonomic than requiring explicit stream binding.
• Consistent use of context.WithTimeout throughout replaces the old per-call MaxWait option.
• Test fix in kafka2jetstream_test.go: switching from tbs.JS.Subscribe to tbs.NC.Subscribe is correct — a plain NATS subscription is simpler and sufficient since JetStream publishes also flow to core NATS subscribers.

[philpennock]

I don't see any dependency updates in here, which is what the PR title says should be here?

The code changes look fine.

[Jarema]

@philpennock I found few issues and addressed them.
Changes were big enough to require a new review.

[piotrpio]

LGTM!

[philpennock]

Imagine a minor grumbling noise about using %s instead of %q for arbitrary subjects logging, combined with "I guess it's consistent with the rest of the code-base, bleh". :)