From b86eb79e99d92bb788bf1629a346355dda054d48 Mon Sep 17 00:00:00 2001 From: scottf Date: Mon, 7 Sep 2026 16:23:24 -0400 Subject: [PATCH 1/2] Support ADR 50 Fast Ingest with constants --- .../support/NatsJetStreamConstants.java | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/src/main/java/io/nats/client/support/NatsJetStreamConstants.java b/src/main/java/io/nats/client/support/NatsJetStreamConstants.java index 7b1adbef6..2d81f5c51 100644 --- a/src/main/java/io/nats/client/support/NatsJetStreamConstants.java +++ b/src/main/java/io/nats/client/support/NatsJetStreamConstants.java @@ -154,10 +154,55 @@ public interface NatsJetStreamConstants { String NATS_BATCH_SEQUENCE_HDR = "Nats-Batch-Sequence"; String NATS_BATCH_COMMIT_HDR = "Nats-Batch-Commit"; + // Values for NATS_BATCH_COMMIT_HDR. + // STORE commits the batch and stores the final message. + // EOB commits the batch without storing the final message. Server 2.14+ + String NATS_BATCH_COMMIT_STORE = "1"; + String NATS_BATCH_COMMIT_EOB = "eob"; + + // Fast ingest batch publish reply subject, which carries the batch control state. + // ......$FI + // The server parses this right to left, so the prefix may itself contain dots. + String FAST_BATCH_SUFFIX = "$FI"; + String FAST_BATCH_GAP_OK = "ok"; + String FAST_BATCH_GAP_FAIL = "fail"; + + // Fast ingest batch publish operations, the subject token just before FAST_BATCH_SUFFIX. + // These are subject tokens, not header values, so they are strings on the wire. + String FAST_BATCH_OP_START = "0"; + String FAST_BATCH_OP_APPEND = "1"; + String FAST_BATCH_OP_COMMIT = "2"; + String FAST_BATCH_OP_COMMIT_EOB = "3"; + String FAST_BATCH_OP_PING = "4"; + + // Fast ingest flow control message types, the value of the "type" field. + // A publish ack has no "type" field, which is what tells the two apart. + String FAST_BATCH_TYPE_ACK = "ack"; + String FAST_BATCH_TYPE_GAP = "gap"; + String FAST_BATCH_TYPE_ERR = "err"; + String NATS_PIN_ID_HDR = "Nats-Pin-Id"; int JS_CONSUMER_NOT_FOUND_ERR = 10014; int JS_NO_MESSAGE_FOUND_ERR = 10037; int JS_WRONG_LAST_SEQUENCE = 10071; int JS_SEQUENCE_TEMPORARILY_UNKNOWN = 10164; + + // Atomic batch publish server errors + int JS_ATOMIC_PUBLISH_DISABLED = 10174; + int JS_ATOMIC_PUBLISH_MISSING_SEQ = 10175; + int JS_ATOMIC_PUBLISH_INCOMPLETE_BATCH = 10176; + int JS_ATOMIC_PUBLISH_UNSUPPORTED_HEADER = 10177; + int JS_ATOMIC_PUBLISH_INVALID_BATCH_ID = 10179; + int JS_ATOMIC_PUBLISH_TOO_LARGE_BATCH = 10199; + int JS_ATOMIC_PUBLISH_INVALID_BATCH_COMMIT = 10200; + int JS_ATOMIC_PUBLISH_DUPLICATE_MESSAGE = 10201; + int JS_ATOMIC_PUBLISH_TOO_MANY_INFLIGHT = 10210; + + // Fast ingest batch publish server errors + int JS_BATCH_PUBLISH_DISABLED = 10205; + int JS_BATCH_PUBLISH_INVALID_PATTERN = 10206; + int JS_BATCH_PUBLISH_INVALID_BATCH_ID = 10207; + int JS_BATCH_PUBLISH_UNKNOWN_BATCH_ID = 10208; + int JS_BATCH_PUBLISH_TOO_MANY_INFLIGHT = 10211; } From ed8d4e3809077333de5ff06609acf5020250e010 Mon Sep 17 00:00:00 2001 From: scottf Date: Mon, 7 Sep 2026 17:18:12 -0400 Subject: [PATCH 2/2] address review --- .../java/io/nats/client/support/NatsJetStreamConstants.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/main/java/io/nats/client/support/NatsJetStreamConstants.java b/src/main/java/io/nats/client/support/NatsJetStreamConstants.java index 2d81f5c51..8621ede21 100644 --- a/src/main/java/io/nats/client/support/NatsJetStreamConstants.java +++ b/src/main/java/io/nats/client/support/NatsJetStreamConstants.java @@ -155,7 +155,8 @@ public interface NatsJetStreamConstants { String NATS_BATCH_COMMIT_HDR = "Nats-Batch-Commit"; // Values for NATS_BATCH_COMMIT_HDR. - // STORE commits the batch and stores the final message. + // Presence of the header marks the commit message, the value selects the mode. + // STORE is the original 2.12 value, a boolean true, and stores the final message. // EOB commits the batch without storing the final message. Server 2.14+ String NATS_BATCH_COMMIT_STORE = "1"; String NATS_BATCH_COMMIT_EOB = "eob"; @@ -194,6 +195,7 @@ public interface NatsJetStreamConstants { int JS_ATOMIC_PUBLISH_INCOMPLETE_BATCH = 10176; int JS_ATOMIC_PUBLISH_UNSUPPORTED_HEADER = 10177; int JS_ATOMIC_PUBLISH_INVALID_BATCH_ID = 10179; + int JS_MIRROR_WITH_ATOMIC_PUBLISH = 10198; int JS_ATOMIC_PUBLISH_TOO_LARGE_BATCH = 10199; int JS_ATOMIC_PUBLISH_INVALID_BATCH_COMMIT = 10200; int JS_ATOMIC_PUBLISH_DUPLICATE_MESSAGE = 10201; @@ -204,5 +206,6 @@ public interface NatsJetStreamConstants { int JS_BATCH_PUBLISH_INVALID_PATTERN = 10206; int JS_BATCH_PUBLISH_INVALID_BATCH_ID = 10207; int JS_BATCH_PUBLISH_UNKNOWN_BATCH_ID = 10208; + int JS_MIRROR_WITH_BATCH_PUBLISH = 10209; int JS_BATCH_PUBLISH_TOO_MANY_INFLIGHT = 10211; }