Make MetricJsonSerializer thread-safe

- Fix the unsafe use of `DateFormatter`

Fix #922

spring-cloud-stream-metrics/src/main/java/org/springframework/cloud/stream/metrics/config/MetricJsonSerializer.java

@@ -22,6 +22,8 @@ import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.TimeZone;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
@@ -42,11 +44,7 @@ import org.springframework.boot.jackson.JsonComponent;
 @JsonComponent
 public class MetricJsonSerializer {
 
-	private final static DateFormat df;
+	private static final BlockingQueue<DateFormat> formatters = new LinkedBlockingQueue<DateFormat>();
-	static {
-		df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSX");
-		df.setTimeZone(TimeZone.getTimeZone("GMT"));
-	}
 
 	public static class Serializer extends JsonSerializer<Metric<?>> {
 
@@ -56,8 +54,17 @@ public class MetricJsonSerializer {
 			json.writeStartObject();
 			json.writeStringField("name", metric.getName());
 			json.writeNumberField("value", metric.getValue().doubleValue());
-			json.writeStringField("timestamp", df.format(metric.getTimestamp()));
+			DateFormat df = formatters.poll();
-			json.writeEndObject();
+			if (df == null) {
+				df = defaultDateFormat();
+			}
+			try {
+				json.writeStringField("timestamp", df.format(metric.getTimestamp()));
+				json.writeEndObject();
+			}
+			finally {
+				formatters.offer(df);
+			}
 		}
 	}
 
@@ -70,16 +77,28 @@ public class MetricJsonSerializer {
 			String name = node.get("name").asText();
 			Number value = node.get("value").asDouble();
 			Date timestamp = null;
-
+			DateFormat df = formatters.poll();
+			if (df == null) {
+				df = defaultDateFormat();
+			}
 			try {
 				timestamp = df.parse(node.get("timestamp").asText());
 			}
 			catch (ParseException e) {
+				// ignore timestamp parsing errors
 			}
-			Metric<Number> metric = new Metric<Number>(name, value, timestamp);
+			finally {
-
+				formatters.offer(df);
+			}
+			Metric<Number> metric = new Metric<>(name, value, timestamp);
 			return metric;
 		}
 
 	}
+
+	private static DateFormat defaultDateFormat() {
+		DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSX");
+		df.setTimeZone(TimeZone.getTimeZone("GMT"));
+		return df;
+	}
 }