debezium/dbz#1581 Add configurable stream name for NATS JetStream sink

Signed-off-by: Ian Muge <[email protected]>

Author: ianmuge

debezium-server-nats-jetstream/src/main/java/io/debezium/server/nats/jetstream/NatsJetStreamChangeConsumer.java

@@ -69,6 +69,7 @@ public class NatsJetStreamChangeConsumer extends BaseChangeConsumer
 
     private static final String PROP_PREFIX = "debezium.sink.nats-jetstream.";
     private static final String PROP_URL = PROP_PREFIX + "url";
+    private static final String PROP_STREAM_NAME = PROP_PREFIX + "stream-name";
     private static final String PROP_CREATE_STREAM = PROP_PREFIX + "create-stream";
     private static final String PROP_SUBJECTS = PROP_PREFIX + "subjects";
     private static final String PROP_STORAGE = PROP_PREFIX + "storage";
@@ -93,6 +94,9 @@ public class NatsJetStreamChangeConsumer extends BaseChangeConsumer
     private JetStream js;
     private RetryExecutor retryExecutor;
 
+    @ConfigProperty(name = PROP_STREAM_NAME, defaultValue = "DebeziumStream")
+    String streamName;
+
     @ConfigProperty(name = PROP_CREATE_STREAM, defaultValue = "false")
     boolean createStream;
 
@@ -179,7 +183,7 @@ else if (tlsKeyStore.isPresent() && tlsKeyStorePassword.isPresent() && tlsPasswo
                 StorageType storageType = storage.equals("file") ? StorageType.File : StorageType.Memory;
 
                 StreamConfiguration streamConfig = StreamConfiguration.builder()
-                        .name("DebeziumStream")
+                        .name(streamName)
                         .description("The debezium stream, contains messages which are coming from debezium")
                         .subjects(subjects.split(","))
                         .storageType(storageType)

debezium-server-nats-jetstream/src/test/java/io/debezium/server/nats/jetstream/NatsJetStreamTestConfigSource.java

@@ -19,6 +19,7 @@ public NatsJetStreamTestConfigSource() {
         natsJetStreamTest.put("debezium.sink.type", "nats-jetstream");
         natsJetStreamTest.put("debezium.sink.nats-jetstream.url",
                 NatsJetStreamTestResourceLifecycleManager.getNatsContainerUrl());
+        natsJetStreamTest.put("debezium.sink.nats-jetstream.stream-name", "DebeziumStream");
         natsJetStreamTest.put("debezium.sink.nats-jetstream.create-stream", "true");
         natsJetStreamTest.put("debezium.sink.nats-jetstream.subjects", "testc.inventory.customers");
         natsJetStreamTest.put("debezium.source.connector.class", "io.debezium.connector.postgresql.PostgresConnector");