Switch the default mode of SUBSTRING to be the 'legacy' mode

docs/syntax-reference.rst

@@ -1099,7 +1099,7 @@ Scalar functions
 |                        |                                                            | It is possible to switch back to this legacy mode |
 |                        |                                                            | by setting                                        |
 |                        |                                                            | ``ksql.functions.substring.legacy.args`` to       |
-|                        |                                                            | ``true``                                          |
+|                        |                                                            | ``true``. Though this is not recommended.         |
 +------------------------+------------------------------------------------------------+---------------------------------------------------+
 | TIMESTAMPTOSTRING      |  ``TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss.SSS')`` | Converts a BIGINT millisecond timestamp value into|
 |                        |                                                            | the string representation of the timestamp in     |

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -262,12 +262,13 @@ private static ConfigDef configDef(final boolean current) {
                + " calling System.exit or executing processes"
         )
         .withClientSslSupport();
-    for (final CompatibilityBreakingConfigDef compatiblityConfigDef
+
+    for (final CompatibilityBreakingConfigDef compatibilityBreakingConfigDef
         : COMPATIBLY_BREAKING_CONFIG_DEBS) {
       if (current) {
-        compatiblityConfigDef.defineCurrent(configDef);
+        compatibilityBreakingConfigDef.defineCurrent(configDef);
       } else {
-        compatiblityConfigDef.defineOld(configDef);
+        compatibilityBreakingConfigDef.defineOld(configDef);
       }
     }
     return configDef;

ksql-engine/src/main/java/io/confluent/ksql/function/udf/string/Substring.java

@@ -17,7 +17,6 @@
 package io.confluent.ksql.function.udf.string;
 
 import io.confluent.common.Configurable;
-import io.confluent.common.config.ConfigException;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -40,8 +39,9 @@ public class Substring implements Configurable {
 
   @Override
   public void configure(final Map<String, ?> props) {
+    final KsqlConfig config = new KsqlConfig(props);
     final boolean legacyArgs =
-        getProps(props, KsqlConfig.KSQL_FUNCTIONS_SUBSTRING_LEGACY_ARGS_CONFIG, false);
+        config.getBoolean(KsqlConfig.KSQL_FUNCTIONS_SUBSTRING_LEGACY_ARGS_CONFIG);
 
     impl = legacyArgs ? new LegacyImpl() : new CurrentImpl();
   }
@@ -121,26 +121,4 @@ private static int getEndIndex(final String value, final int start, final int le
       return Math.max(Math.min(start + length, value.length()), start);
     }
   }
-
-  @SuppressWarnings("SameParameterValue")
-  private static boolean getProps(
-      final Map<String,?> props,
-      final String name,
-      final boolean defaultValue) {
-
-    final Object value = props.get(name);
-    if (value == null) {
-      return defaultValue;
-    }
-
-    if (value instanceof String) {
-      return Boolean.valueOf((String)value);
-    }
-
-    if (!(value instanceof Boolean)) {
-      throw new ConfigException(name, value, "Value is not boolean");
-    }
-
-    return (Boolean)value;
-  }
 }

ksql-engine/src/test/java/io/confluent/ksql/function/udf/string/SubstringTest.java

@@ -21,8 +21,8 @@
 import static org.junit.Assert.assertThat;
 
 import com.google.common.collect.ImmutableMap;
-import io.confluent.common.config.ConfigException;
 import io.confluent.ksql.util.KsqlConfig;
+import org.apache.kafka.common.config.ConfigException;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -215,15 +215,6 @@ public void shouldNotEnterLegacyModeWithFalseBooleanConfig() {
     assertThat(udfIsInLegacyMode(), is(false));
   }
 
-  @Test
-  public void shouldNotEnterLegacyModeWithOtherStringConfig() {
-    // When:
-    configure("what ever");
-
-    // Then:
-    assertThat(udfIsInLegacyMode(), is(false));
-  }
-
   @Test(expected = ConfigException.class)
   public void shouldThrowOnInvalidLegacyModeValueType() {
     configure(1.0);

ksql-engine/src/test/resources/query-validation-tests/substring.json

@@ -1,11 +1,17 @@
 {
   "comments": [
-    "Tests covering the use of the SUBSTRING function"
+    "Tests covering the use of the SUBSTRING function.",
+    "NOTE: At present the SUBSTRING function has two modes: current and legacy.",
+    " v5.0 and earlier only have legacy mode.",
+    " v5.1 and later have both current and legacy. New queries default to the current mode"
   ],
   "tests": [
     {
       "name": "do substring with just pos",
       "format": ["JSON"],
+      "properties": {
+        "ksql.functions.substring.legacy.args": false
+      },
       "statements": [
         "CREATE STREAM TEST (source VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON');",
         "CREATE STREAM OUTPUT AS SELECT SUBSTRING(source, 6) AS SUBSTRING, SUBSTRING(null, 1) AS NULL_STR, SUBSTRING(source, null) AS NULL_POS FROM TEST;"
@@ -24,7 +30,9 @@
     {
       "name": "do substring with pos and length",
       "format": ["JSON"],
-      "statements": [
+      "properties": {
+        "ksql.functions.substring.legacy.args": false
+      },"statements": [
         "CREATE STREAM TEST (source VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON');",
         "CREATE STREAM OUTPUT AS SELECT SUBSTRING(source, 6, 3) AS SUBSTRING, SUBSTRING(null, 1) AS NULL_STR, SUBSTRING(source, null) AS NULL_POS, SUBSTRING(source, 6, null) AS NULL_LEN FROM TEST;"
       ],
@@ -80,6 +88,20 @@
         {"topic": "OUTPUT", "key": 2, "value": {"SUBSTRING":"nothe", "NULL_STR":null, "NULL_START":null, "NULL_END":null}, "timestamp": 0},
         {"topic": "OUTPUT", "key": 2, "value": {"SUBSTRING":null, "NULL_STR":null, "NULL_START":null, "NULL_END":null}, "timestamp": 0}
       ]
+    },
+    {
+      "name": "should default to current mode for new queries",
+      "format": ["JSON"],
+      "statements": [
+        "CREATE STREAM TEST (source VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON');",
+        "CREATE STREAM OUTPUT AS SELECT SUBSTRING(source, 6) AS SUBSTRING FROM TEST;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "key": 1, "value": {"source": "some_string"}, "timestamp": 0}
+      ],
+      "outputs": [
+        {"topic": "OUTPUT", "key": 1, "value": {"SUBSTRING":"string"}, "timestamp": 0}
+      ]
     }
   ]
 }