DBZ-7829 Added nats JWT/seed authentication config options

Added two new properties; `auth.jwt` and `auth.seed` to allow to pass a JWT and a seed to connect to nats when it has been configured for decentralized auth (like Synadia) Signed-off-by: Daan Gerits <[email protected]>
debezium · Apr 29, 2024 · df13b31 · df13b31
1 parent 11c0770
commit df13b31
Showing 1 changed file with 18 additions and 4 deletions.
diff --git a/...etstream/src/main/java/io/debezium/server/nats/jetstream/NatsJetStreamChangeConsumer.java b/...etstream/src/main/java/io/debezium/server/nats/jetstream/NatsJetStreamChangeConsumer.java
@@ -30,6 +30,7 @@
 import io.nats.client.JetStream;
 import io.nats.client.JetStreamManagement;
 import io.nats.client.Nats;
+import io.nats.client.Options;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
 
@@ -51,12 +52,21 @@ public class NatsJetStreamChangeConsumer extends BaseChangeConsumer
     private static final String PROP_SUBJECTS = PROP_PREFIX + "subjects";
     private static final String PROP_STORAGE = PROP_PREFIX + "storage";
 
+    private static final String PROP_AUTH_JWT = PROP_PREFIX + "auth.jwt";
+    private static final String PROP_AUTH_SEED = PROP_PREFIX + "auth.seed";
+
     private Connection nc;
     private JetStream js;
 
     @ConfigProperty(name = PROP_CREATE_STREAM, defaultValue = "false")
     boolean createStream;
 
+    @ConfigProperty(name = PROP_AUTH_JWT, defaultValue = "")
+    String jwt;
+
+    @ConfigProperty(name = PROP_AUTH_SEED, defaultValue = "")
+    String seed;
+
     @Inject
     @CustomConsumerBuilder
     Instance<JetStream> customStreamingConnection;
@@ -75,11 +85,15 @@ void connect() {
 
         try {
             // Setup NATS connection
-            io.nats.client.Options natsOptions = new io.nats.client.Options.Builder()
+            Options.Builder natsOptionsBuilder = new io.nats.client.Options.Builder()
                     .servers(url.split(","))
-                    .noReconnect()
-                    .build();
-            nc = Nats.connect(natsOptions);
+                    .noReconnect();
+
+            if (!jwt.isEmpty() && !seed.isEmpty()) {
+                natsOptionsBuilder.authHandler(Nats.staticCredentials(jwt.toCharArray(), seed.toCharArray()));
+            }
+
+            nc = Nats.connect(natsOptionsBuilder.build());
 
             // Creating a basic stream, mostly for testing. If a user wants to configure their stream, it can be done
             // via the nats cli.