fix TimeDimExtractionFn.apply() under concurrency (#3984)

processing/src/main/java/io/druid/query/extraction/ExtractionFn.java

@@ -46,7 +46,8 @@ import io.druid.query.lookup.RegisteredLookupExtractionFn;
     @JsonSubTypes.Type(name = "strlen", value = StrlenExtractionFn.class)
 })
 /**
- * An ExtractionFn is a function that can be used to transform the values of a column (typically a dimension)
+ * An ExtractionFn is a function that can be used to transform the values of a column (typically a dimension).
+ * Note that ExtractionFn implementations are expected to be Threadsafe.
  *
  * A simple example of the type of operation this enables is the RegexDimExtractionFn which applies a
  * regular expression with a capture group.  When the regular expression matches the value of a dimension,

processing/src/main/java/io/druid/query/extraction/TimeDimExtractionFn.java

@@ -34,9 +34,9 @@ import java.util.Date;
 public class TimeDimExtractionFn extends DimExtractionFn
 {
   private final String timeFormat;
-  private final SimpleDateFormat timeFormatter;
+  private final ThreadLocal<SimpleDateFormat> timeFormatter;
   private final String resultFormat;
-  private final SimpleDateFormat resultFormatter;
+  private final ThreadLocal<SimpleDateFormat> resultFormatter;
 
   @JsonCreator
   public TimeDimExtractionFn(
@@ -48,11 +48,23 @@ public class TimeDimExtractionFn extends DimExtractionFn
     Preconditions.checkNotNull(resultFormat, "resultFormat must not be null");
 
     this.timeFormat = timeFormat;
-    this.timeFormatter = new SimpleDateFormat(timeFormat);
-    this.timeFormatter.setLenient(true);
+    this.timeFormatter = new ThreadLocal<SimpleDateFormat>() {
+      @Override
+      public SimpleDateFormat initialValue() {
+        SimpleDateFormat formatter = new SimpleDateFormat(TimeDimExtractionFn.this.timeFormat);
+        formatter.setLenient(true);
+        return formatter;
+      }
+    };
 
     this.resultFormat = resultFormat;
-    this.resultFormatter = new SimpleDateFormat(resultFormat);
+    this.resultFormatter = new ThreadLocal<SimpleDateFormat>() {
+      @Override
+      public SimpleDateFormat initialValue() {
+        SimpleDateFormat formatter = new SimpleDateFormat(TimeDimExtractionFn.this.resultFormat);
+        return formatter;
+      }
+    };
   }
 
   @Override
@@ -70,12 +82,12 @@ public class TimeDimExtractionFn extends DimExtractionFn
   {
     Date date;
     try {
-      date = timeFormatter.parse(dimValue);
+      date = timeFormatter.get().parse(dimValue);
     }
     catch (ParseException e) {
       return dimValue;
     }
-    return resultFormatter.format(date);
+    return resultFormatter.get().format(date);
   }
 
   @JsonProperty("timeFormat")