Add nats_creds setting for using JWT auth

docs/docs/config.md

@@ -106,7 +106,8 @@ _See also:_ [SNS Producer Documentation](/producers#sns)
 option                         | argument                            | description                                         | default
 -------------------------------|-------------------------------------| --------------------------------------------------- | -------
 nats_url                       | STRING     | Comma separated list of nats urls.  may include [user:password style auth](https://docs.nats.io/developing-with-nats/security/userpass#connecting-with-a-user-password-in-the-url) | nats://localhost:4222
-nats_subject                   | STRING     | Nats subject hierarchy.  [Topic substitution](/producers/#topic-substitution) available. | `%{database}.%{table}`
+nats_subject                   | STRING     | Nats subject hierarchy.  [Topic substitution](/producers/#topic-substitution) available.| `%{database}.%{table}`
+nats_creds                     | STRING     | [Nats creds file path](https://docs.nats.io/using-nats/developer/connecting/creds) for JWT + NKey auth.| null
 
 _See also:_ [Nats Producer Documentation](/producers#nats)
 

docs/docs/producers.md

@@ -247,10 +247,14 @@ The configurable properties for nats are:
 
 - `nats_url` - defaults to **nats://localhost:4222**
 - `nats_subject` - defaults to **%{database}.%{table}**
+- `nats_creds` - defaults to null
 
 `nats_subject` defines the Nats subject hierarchy to write to.  [Topic substitution](/producers#topic-substitution) is available.
 All non-alphanumeric characters in the substitued values will be replaced by underscores.
 
+`nats_creds` is the path to a [NATS .creds file](https://docs.nats.io/using-nats/developer/connecting/creds), which can be generated using the `nsc` tool. 
+It is a text file containing the NATS user JWT and NKEY for use with Decentralized Auth.
+
 # Google Cloud Pub/Sub
 ***
 In order to publish to Google Cloud Pub/Sub, you will need to obtain an IAM service account that has been granted the `roles/pubsub.publisher` role.

pom.xml

@@ -295,7 +295,7 @@
     <dependency>
       <groupId>io.nats</groupId>
       <artifactId>jnats</artifactId>
-      <version>2.8.0</version>
+      <version>2.17.1</version>
     </dependency>
     <dependency>
       <groupId>com.google.protobuf</groupId>

src/main/java/com/zendesk/maxwell/MaxwellConfig.java

@@ -556,6 +556,11 @@ public class MaxwellConfig extends AbstractConfig {
 	 */
 	public String natsSubject;
 
+	/**
+	 * {@link com.zendesk.maxwell.producer.NatsProducer} Credential file
+	 */
+	public String natsCreds;
+
 	/**
 	 * {@link com.zendesk.maxwell.producer.MaxwellRedisProducer} host
 	 */
@@ -898,6 +903,7 @@ protected MaxwellOptionParser buildOptionParser() {
 
 		parser.accepts( "nats_url", "Url(s) of Nats connection (comma separated). Default is localhost:4222" ).withRequiredArg();
 		parser.accepts( "nats_subject", "Subject Hierarchies of Nats. Default is '%{database}.%{table}'" ).withRequiredArg();
+		parser.accepts( "nats_creds", "Nats creds file path for JWT + NKey auth. Default is null which disables JWT based auth" ).withRequiredArg();
 
 		parser.section( "bigquery" );
 		parser.accepts( "bigquery_project_id", "provide a google cloud platform project id associated with the bigquery table" )
@@ -1110,6 +1116,7 @@ private void setup(OptionSet options, Properties properties) {
 
 		this.natsUrl			= fetchStringOption("nats_url", options, properties, "nats://localhost:4222");
 		this.natsSubject		= fetchStringOption("nats_subject", options, properties, "%{database}.%{table}");
+		this.natsCreds			= fetchStringOption("nats_creds", options, properties, null);
 
 		this.redisHost			= fetchStringOption("redis_host", options, properties, "localhost");
 		this.redisPort			= fetchIntegerOption("redis_port", options, properties, 6379);

src/main/java/com/zendesk/maxwell/producer/NatsProducer.java

@@ -25,6 +25,10 @@ public NatsProducer(MaxwellContext context) {
 		List<String> urls = Arrays.asList(context.getConfig().natsUrl.split(","));
 		Options.Builder optionBuilder = new Options.Builder();
 		urls.forEach(optionBuilder::server);
+		String credsFile = context.getConfig().natsCreds;
+		if (credsFile != null) {
+			optionBuilder.authHandler(Nats.credentials(credsFile));
+		}
 		Options option = optionBuilder.build();
 
 		this.natsSubjectTemplate = context.getConfig().natsSubject;