fix: Allow multiple EXTRACTJSONFIELD invocations on different paths (#…

…8122) (#8123)
confluentinc · Sep 21, 2021 · 7f1d407 · 7f1d407
1 parent cc5c02b
commit 7f1d407
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 4 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/json/JsonExtractString.java b/ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/json/JsonExtractString.java
@@ -37,7 +37,8 @@ public class JsonExtractString {
 
   private static final ObjectReader OBJECT_READER = UdfJsonMapper.INSTANCE.get().reader();
 
-  private List<String> tokens = null;
+  private String latestPath = null;
+  private List<String> latestTokens = null;
 
   @Udf
   public String extract(
@@ -48,13 +49,15 @@ public String extract(
       return null;
     }
 
-    if (tokens == null) {
+
+    if (latestPath == null || !latestPath.equals(path)) {
       final JsonPathTokenizer tokenizer = new JsonPathTokenizer(path);
-      tokens = ImmutableList.copyOf(tokenizer);
+      latestTokens = ImmutableList.copyOf(tokenizer);
+      latestPath = path;
     }
 
     JsonNode currentNode = parseJsonDoc(input);
-    for (final String token : tokens) {
+    for (final String token : latestTokens) {
       if (currentNode instanceof ArrayNode) {
         try {
           final int index = Integer.parseInt(token);

diff --git a/...b-engine/src/test/java/io/confluent/ksql/function/udf/json/JsonExtractStringKudfTest.java b/...b-engine/src/test/java/io/confluent/ksql/function/udf/json/JsonExtractStringKudfTest.java
@@ -105,4 +105,13 @@ public void shouldBeThreadSafe() {
         .parallel()
         .forEach(idx -> shouldExtractJsonField());
   }
+
+  @Test
+  public void shouldParseCorrectlyDifferentPathsOnSameInstance() {
+    final String thing1 = udf.extract(JSON_DOC, "$.thing1");
+    assertThat(thing1, is("{\"thing2\":\"hello\"}"));
+
+    final String array = udf.extract(JSON_DOC, "$.array.1");
+    assertThat(array, is("102"));
+  }
 }