Merge pull request #1301 from nats-io/kv-max-age-vs-duplicate-window

KV TTL (stream max_age) versus stream duplicate_window

Author: scottf

src/main/java/io/nats/client/api/FeatureConfiguration.java

@@ -137,8 +137,9 @@ public Map<String, String> getMetadata() {
     }
 
     protected static abstract class Builder<B, FC> {
-        String name;
-        StreamConfiguration.Builder scBuilder;
+        protected String name;
+        protected Duration ttl = Duration.ZERO;
+        protected StreamConfiguration.Builder scBuilder;
         protected abstract B getThis();
 
         /**
@@ -177,7 +178,8 @@ protected B maxBucketSize(long maxBucketSize) {
          * @return Builder
          */
         protected B ttl(Duration ttl) {
-            scBuilder.maxAge(ttl);
+            this.ttl = ttl == null ? Duration.ZERO : ttl;
+            scBuilder.maxAge(this.ttl);
             return getThis();
         }
 

src/main/java/io/nats/client/api/KeyValueConfiguration.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.*;
 
+import static io.nats.client.support.NatsJetStreamConstants.SERVER_DEFAULT_DUPLICATE_WINDOW_MS;
 import static io.nats.client.support.NatsKeyValueUtil.*;
 import static io.nats.client.support.Validator.*;
 
@@ -420,6 +421,18 @@ else if (!sources.isEmpty()) {
                 scBuilder.subjects(toStreamSubject(name));
             }
 
+            // When stream's MaxAge is not set, server uses 2 minutes as the default
+            // for the duplicate window. If MaxAge is set, and lower than 2 minutes,
+            // then the duplicate window will be set to that. If MaxAge is greater,
+            // we will cap the duplicate window to 2 minutes (to be consistent with
+            // previous behavior).
+            long ttlMs = ttl.toMillis();
+            long dupeMs = SERVER_DEFAULT_DUPLICATE_WINDOW_MS;
+            if (ttlMs > 0 && ttlMs < SERVER_DEFAULT_DUPLICATE_WINDOW_MS) {
+                dupeMs = ttlMs;
+            }
+            scBuilder.duplicateWindow(dupeMs);
+
             return new KeyValueConfiguration(scBuilder.build());
         }
     }

src/main/java/io/nats/client/support/NatsJetStreamConstants.java

@@ -12,6 +12,8 @@ public interface NatsJetStreamConstants {
      */
     int MAX_HISTORY_PER_KEY = 64;
 
+    long SERVER_DEFAULT_DUPLICATE_WINDOW_MS = 120_000; // 1000ms/sec * 60sec/min * 2 min
+
     String PREFIX_DOLLAR_JS_DOT = "$JS.";
     String PREFIX_API = "API";
     String DEFAULT_API_PREFIX = "$JS.API.";

src/test/java/io/nats/client/impl/KeyValueTests.java

@@ -30,6 +30,7 @@
 import static io.nats.client.api.KeyValuePurgeOptions.DEFAULT_THRESHOLD_MILLIS;
 import static io.nats.client.api.KeyValueWatchOption.*;
 import static io.nats.client.support.NatsConstants.DOT;
+import static io.nats.client.support.NatsJetStreamConstants.SERVER_DEFAULT_DUPLICATE_WINDOW_MS;
 import static org.junit.jupiter.api.Assertions.*;
 
 public class KeyValueTests extends JetStreamTestBase {
@@ -1709,5 +1710,40 @@ public void testSubjectFiltersAgainst209OptOut() throws Exception {
             assertKeys(kv.keys(Arrays.asList("one", "two")), "one", "two");
         });
     }
+
+    @Test
+    public void testTtlAndDuplicateWindow() throws Exception {
+        jsServer.run(TestBase::atLeast2_10, nc -> {
+            KeyValueManagement kvm = nc.keyValueManagement();
+            String bucket = bucket();
+            KeyValueConfiguration config = KeyValueConfiguration.builder()
+                .name(bucket)
+                .storageType(StorageType.Memory)
+                .build();
+            KeyValueStatus status = kvm.create(config);
+
+            StreamConfiguration sc = status.getBackingStreamInfo().getConfiguration();
+            assertEquals(0, sc.getMaxAge().toMillis());
+            assertEquals(SERVER_DEFAULT_DUPLICATE_WINDOW_MS, sc.getDuplicateWindow().toMillis());
+
+            config = KeyValueConfiguration.builder(status.getConfiguration()).ttl(Duration.ofSeconds(10)).build();
+            status = kvm.update(config);
+            sc = status.getBackingStreamInfo().getConfiguration();
+            assertEquals(10_000, sc.getMaxAge().toMillis());
+            assertEquals(10_000, sc.getDuplicateWindow().toMillis());
+
+            bucket = bucket();
+            config = KeyValueConfiguration.builder()
+                .name(bucket)
+                .storageType(StorageType.Memory)
+                .ttl(Duration.ofMinutes(30))
+                .build();
+            status = kvm.create(config);
+
+            sc = status.getBackingStreamInfo().getConfiguration();
+            assertEquals(30, sc.getMaxAge().toMinutes());
+            assertEquals(SERVER_DEFAULT_DUPLICATE_WINDOW_MS, sc.getDuplicateWindow().toMillis());
+        });
+    }
 }