This PR adds a production-ready NATS JetStream connector to Proton, bringing it to full feature parity with the existing Kafka external stream. The connector supports pull-based subscriptions with configurable batch sizes, durable consumers that survive restarts, and three acknowledgement policies (none, all, and explicit) for at-least-once delivery semantics. On the authentication side it covers all five methods NATS supports — username/password, token, NKey, credentials file, and mutual TLS — with TLS being auto-detected from the connection URL scheme.The write path supports dynamic subject routing via _tp_message_key and full header propagation through _tp_message_headers as a Map(String, String). The read path includes stall detection that automatically recreates the subscription if no progress is made within a configurable timeout, and checkpointing serializes the subject, consumer name, and last sequence number to the existing CheckpointCoordinator so materialized views recover correctly.A big thanks to @yuzifeng1984 for porting the nats.c library as a submodule and fixing several compilation issues: renaming the guard macro to USE_NATSIO for consistency with config.h.in, replacing natsMsg_GetTimestamp() (which doesn't exist in the API) with natsMsg_GetMetaData, correcting the jsFetchRequest timeout field from MaxWait to Expires, and replacing the non-existent nats_Free() with a plain free() since natsMsgHeader_Keys allocates a standard C array. Note: one_message_per_row is intentionally inherited from the shared external stream settings already defined in Kafka — no separate definition is needed for NATS.

Closes #535

NATS JetStream External Stream — Feature Reference

Overview

The NATS JetStream external stream allows Proton to read from and write to NATS JetStream — NATS's built-in persistence layer. It follows the same external stream pattern as Kafka, supporting streaming reads, historical reads, INSERTs, materialized views, and checkpointing.

Prerequisites

• A NATS server with JetStream enabled (nats-server --jetstream)
• The JetStream stream must already exist on the NATS server before creating the external stream in Proton (Proton validates this on CREATE)

CREATE EXTERNAL STREAM Syntax

CREATE EXTERNAL STREAM stream_name (
    col1 type1,
    col2 type2,
    ...
)
SETTINGS
    type            = 'nats_jetstream',
    url             = 'nats://host:4222',
    subject         = 'orders.>',
    stream_name     = 'ORDERS',
    data_format     = 'JSONEachRow';

Required settings:

Format settings:

Consumer settings:

Authentication (use at most one):

TLS:

Reconnection:

Virtual Columns

Every NATS JetStream external stream automatically exposes these virtual columns (no need to define them in the schema):

SELECT _tp_sn, _nats_subject, _tp_message_headers, order_id, amount
FROM table(orders_stream);

Reading (SELECT)

Historical / bounded read — fetches all existing messages and returns:

SELECT * FROM table(nats_stream);

Streaming / unbounded read — runs continuously, consuming new messages in real time:

SELECT * FROM nats_stream;

Subject filtering — use specific subjects (not wildcards) to read a subset:

-- Only US orders (stream ORDERS captures 'orders.>')
CREATE EXTERNAL STREAM us_orders (...)
SETTINGS type='nats_jetstream', subject='orders.us', stream_name='ORDERS', ...;

SELECT * FROM us_orders;

Writing (INSERT)

Write messages to NATS JetStream via INSERT. The subject setting must be a concrete subject (not a wildcard):

CREATE EXTERNAL STREAM orders_sink (
    order_id uint32, product string, amount uint32, region string
)
SETTINGS
    type = 'nats_jetstream', url = 'nats://localhost:4222',
    subject = 'orders.us', stream_name = 'ORDERS',
    data_format = 'JSONEachRow', one_message_per_row = true;

INSERT INTO orders_sink (order_id, product, amount, region)
VALUES (101, 'widget', 999, 'us');

Per-row subject routing — define _tp_message_key as a column to override the publish subject per row:

CREATE EXTERNAL STREAM orders_routed (
    order_id uint32, product string, region string,
    _tp_message_key string    -- overrides subject per row
)
SETTINGS type='nats_jetstream', subject='orders.us', stream_name='ORDERS',
         data_format='JSONEachRow', one_message_per_row=true;

INSERT INTO orders_routed (order_id, product, region, _tp_message_key)
VALUES (201, 'gadget', 'jp', 'orders.jp'),
       (202, 'widget', 'de', 'orders.de');

Custom NATS headers — define _tp_message_headers to attach headers to each message:

CREATE EXTERNAL STREAM orders_with_headers (
    order_id uint32, product string, amount uint32,
    _tp_message_headers map(string, string)
)
SETTINGS type='nats_jetstream', subject='orders.us', stream_name='ORDERS',
         data_format='JSONEachRow', one_message_per_row=true;

INSERT INTO orders_with_headers (order_id, product, amount, _tp_message_headers)
VALUES (301, 'premium', 5000, {'X-Priority':'high', 'X-Source':'proton'});

Materialized Views (Continuous ETL)

Pipe data from NATS JetStream into a Proton stream continuously:

CREATE STREAM order_archive (
    order_id uint32, product string, amount uint32, region string
);

CREATE MATERIALIZED VIEW orders_etl INTO order_archive AS
    SELECT order_id, product, amount, region
    FROM nats_orders_stream
    WHERE amount > 100;

-- Query the archive anytime:
SELECT * FROM table(order_archive) ORDER BY _tp_time;

The MV:

• Creates a durable consumer on the JetStream stream
• Supports checkpointing (survives Proton restarts)
• Automatically resumes from the last checkpoint position

Deliver Policies

-- Start from sequence 100
CREATE EXTERNAL STREAM orders_from_100 (...)
SETTINGS type='nats_jetstream', ..., deliver_policy='by_start_sequence', start_sequence=100;

-- Start from a specific time (2026-01-01T00:00:00Z in nanoseconds)
CREATE EXTERNAL STREAM orders_from_2026 (...)
SETTINGS type='nats_jetstream', ..., deliver_policy='by_start_time', start_time='1767225600000000000';

Durable Consumers

When durable = true (default), the consumer position is persisted on the NATS server. If the query is restarted, it resumes from where it left off:

CREATE EXTERNAL STREAM orders_durable (...)
SETTINGS type='nats_jetstream', ...,
    durable = true,
    consumer_name = 'my-app-consumer',
    ack_policy = 'explicit';

-- First run: consumes all existing messages
SELECT * FROM orders_durable;

-- Second run: only new messages (position is remembered by NATS)
SELECT * FROM orders_durable;

Comparison with Kafka External Stream

Limitations & Notes

• The JetStream stream must already exist on the NATS server. Proton does not create it.
• Wildcard subjects (orders.>, orders.*) work for reading but not for writing. Use a concrete subject for INSERT.
• _tp_message_key for NATS is always a string (NATS subjects are strings), unlike Kafka where it can be bytes.
• _tp_shard is always 0 since NATS JetStream doesn't have a partition concept.
• When defining _tp_message_key, _tp_time, or _tp_message_headers as physical columns, one_message_per_row is forced to true.