Iteration 3: Addresses sfc-gh-asen's comments

snowflakedb · Sep 13, 2023 · bd716ab · bd716ab
1 parent 6b21fab
commit bd716ab
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 9 deletions.
diff --git a/src/main/java/net/snowflake/ingest/utils/ParameterProvider.java b/src/main/java/net/snowflake/ingest/utils/ParameterProvider.java
@@ -188,29 +188,30 @@ public long getBufferFlushIntervalInMs() {
   private long getMaxClientLagMs() {
     Object val = this.parameterMap.getOrDefault(MAX_CLIENT_LAG, MAX_CLIENT_LAG_DEFAULT);
     if (!(val instanceof String)) {
-      return 1000;
+      return BUFFER_FLUSH_INTERVAL_IN_MILLIS_DEFAULT;
     }
     String maxLag = (String) val;
     String[] lagParts = maxLag.split(" ");
     if (lagParts.length != 2
         || (lagParts[0] == null || "".equals(lagParts[0]))
         || (lagParts[1] == null || "".equals(lagParts[1]))) {
-      return 1000;
+      throw new IllegalArgumentException(
+          String.format("Failed to parse MAX_CLIENT_LAG = '%s'", maxLag));
     }
-    long unit;
+    long lag;
     try {
-      unit = Long.parseLong(lagParts[0]);
+      lag = Long.parseLong(lagParts[0]);
     } catch (Throwable t) {
       throw new IllegalArgumentException(
           String.format("Failed to parse MAX_CLIENT_LAG = '%s'", lagParts[0]), t);
     }
     switch (lagParts[1].toLowerCase()) {
       case "second":
       case "seconds":
-        return unit * 1000;
+        return lag * 1000;
       case "minute":
       case "minutes":
-        return unit * 60000;
+        return lag * 60000;
       default:
         throw new IllegalArgumentException(
             String.format("Invalid time unit supplied = '%s", lagParts[1]));
@@ -220,7 +221,7 @@ private long getMaxClientLagMs() {
   private boolean getMaxClientLagEnabled() {
     Object val =
         this.parameterMap.getOrDefault(MAX_CLIENT_LAG_ENABLED, MAX_CLIENT_LAG_ENABLED_DEFAULT);
-    return (val instanceof Boolean) ? (Boolean) val : false;
+    return (val instanceof String) ? Boolean.parseBoolean(val.toString()) : (boolean) val;
   }
 
   /** @return Time in milliseconds between checks to see if the buffer should be flushed */

diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/ParameterProviderTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/ParameterProviderTest.java
@@ -190,7 +190,12 @@ public void testMaxClientLagEnabledMissingUnit() {
     parameterMap.put(ParameterProvider.MAX_CLIENT_LAG_ENABLED, true);
     parameterMap.put(ParameterProvider.MAX_CLIENT_LAG, "1");
     ParameterProvider parameterProvider = new ParameterProvider(parameterMap, prop);
-    Assert.assertEquals(1000, parameterProvider.getBufferFlushIntervalInMs());
+    try {
+      parameterProvider.getBufferFlushIntervalInMs();
+      Assert.fail("Should not have succeeded");
+    } catch (IllegalArgumentException e) {
+      Assert.assertTrue(e.getMessage().startsWith("Failed to parse"));
+    }
   }
 
   @Test
@@ -200,7 +205,12 @@ public void testMaxClientLagEnabledMissingUnitTimeUnitSupplied() {
     parameterMap.put(ParameterProvider.MAX_CLIENT_LAG_ENABLED, true);
     parameterMap.put(ParameterProvider.MAX_CLIENT_LAG, " year");
     ParameterProvider parameterProvider = new ParameterProvider(parameterMap, prop);
-    Assert.assertEquals(1000, parameterProvider.getBufferFlushIntervalInMs());
+    try {
+      parameterProvider.getBufferFlushIntervalInMs();
+      Assert.fail("Should not have succeeded");
+    } catch (IllegalArgumentException e) {
+      Assert.assertTrue(e.getMessage().startsWith("Failed to parse"));
+    }
   }
 
   @Test