[FLINK-16342][metrics][datadog] Remove mocking

- introduce timestamp factory instead of relying on static method
- introduce static package-private non-final flag for skipping api validation

flink-metrics/flink-metrics-datadog/pom.xml

@@ -33,6 +33,13 @@ under the License.
 	<name>flink-metrics-datadog</name>
 
 	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-annotations</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-metrics-core</artifactId>

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/Clock.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.metrics.datadog;
+
+/**
+ * A simple clock that returns the number of seconds since the unix epoch.
+ */
+public interface Clock {
+	long getUnixEpochTimestamp();
+}

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DCounter.java

@@ -28,8 +28,8 @@ import java.util.List;
 public class DCounter extends DMetric {
 	private final Counter counter;
 
-	public DCounter(Counter c, String metricName, String host, List<String> tags) {
-		super(MetricType.count, metricName, host, tags);
+	public DCounter(Counter c, String metricName, String host, List<String> tags, Clock clock) {
+		super(MetricType.count, metricName, host, tags, clock);
 		counter = c;
 	}
 

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DGauge.java

@@ -28,8 +28,8 @@ import java.util.List;
 public class DGauge extends DMetric {
 	private final Gauge<Number> gauge;
 
-	public DGauge(Gauge<Number> g, String metricName, String host, List<String> tags) {
-		super(MetricType.gauge, metricName, host, tags);
+	public DGauge(Gauge<Number> g, String metricName, String host, List<String> tags, Clock clock) {
+		super(MetricType.gauge, metricName, host, tags, clock);
 		gauge = g;
 	}
 

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMeter.java

@@ -30,8 +30,8 @@ import java.util.List;
 public class DMeter extends DMetric {
 	private final Meter meter;
 
-	public DMeter(Meter m, String metricName, String host, List<String> tags) {
-		super(MetricType.gauge, metricName, host, tags);
+	public DMeter(Meter m, String metricName, String host, List<String> tags, Clock clock) {
+		super(MetricType.gauge, metricName, host, tags, clock);
 		meter = m;
 	}
 

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java

@@ -29,7 +29,6 @@ import java.util.List;
  */
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public abstract class DMetric {
-	private static final long MILLIS_TO_SEC = 1000L;
 
 	/**
 	 * Names of metric/type/tags field and their getters must not be changed
@@ -39,12 +38,14 @@ public abstract class DMetric {
 	private final MetricType type;
 	private final String host;
 	private final List<String> tags;
+	private final Clock clock;
 
-	public DMetric(MetricType metricType, String metric, String host, List<String> tags) {
+	public DMetric(MetricType metricType, String metric, String host, List<String> tags, Clock clock) {
 		this.type = metricType;
 		this.metric = metric;
 		this.host = host;
 		this.tags = tags;
+		this.clock = clock;
 	}
 
 	public MetricType getType() {
@@ -66,7 +67,7 @@ public abstract class DMetric {
 	public List<List<Number>> getPoints() {
 		// One single data point
 		List<Number> point = new ArrayList<>();
-		point.add(getUnixEpochTimestamp());
+		point.add(clock.getUnixEpochTimestamp());
 		point.add(getMetricValue());
 
 		List<List<Number>> points = new ArrayList<>();
@@ -77,8 +78,4 @@ public abstract class DMetric {
 
 	@JsonIgnore
 	public abstract Number getMetricValue();
-
-	public static long getUnixEpochTimestamp() {
-		return (System.currentTimeMillis() / MILLIS_TO_SEC);
-	}
 }

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java

@@ -56,7 +56,7 @@ public class DatadogHttpClient {
 	private final String proxyHost;
 	private final int proxyPort;
 
-	public DatadogHttpClient(String dgApiKey, String dgProxyHost, int dgProxyPort) {
+	public DatadogHttpClient(String dgApiKey, String dgProxyHost, int dgProxyPort, boolean validateApiKey) {
 		if (dgApiKey == null || dgApiKey.isEmpty()) {
 			throw new IllegalArgumentException("Invalid API key:" + dgApiKey);
 		}
@@ -75,7 +75,9 @@ public class DatadogHttpClient {
 
 		seriesUrl = String.format(SERIES_URL_FORMAT, apiKey);
 		validateUrl = String.format(VALIDATE_URL_FORMAT, apiKey);
-		validateApiKey();
+		if (validateApiKey) {
+			validateApiKey();
+		}
 	}
 
 	Proxy getProxy() {

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java

@@ -55,6 +55,8 @@ public class DatadogHttpReporter implements MetricReporter, Scheduled {
 	private DatadogHttpClient client;
 	private List<String> configTags;
 
+	private final Clock clock = () -> System.currentTimeMillis() / 1000L;
+
 	public static final String API_KEY = "apikey";
 	public static final String PROXY_HOST = "proxyHost";
 	public static final String PROXY_PORT = "proxyPort";
@@ -70,14 +72,14 @@ public class DatadogHttpReporter implements MetricReporter, Scheduled {
 
 		if (metric instanceof Counter) {
 			Counter c = (Counter) metric;
-			counters.put(c, new DCounter(c, name, host, tags));
+			counters.put(c, new DCounter(c, name, host, tags, clock));
 		} else if (metric instanceof Gauge) {
 			Gauge g = (Gauge) metric;
-			gauges.put(g, new DGauge(g, name, host, tags));
+			gauges.put(g, new DGauge(g, name, host, tags, clock));
 		} else if (metric instanceof Meter) {
 			Meter m = (Meter) metric;
 			// Only consider rate
-			meters.put(m, new DMeter(m, name, host, tags));
+			meters.put(m, new DMeter(m, name, host, tags, clock));
 		} else if (metric instanceof Histogram) {
 			LOGGER.warn("Cannot add {} because Datadog HTTP API doesn't support Histogram", metricName);
 		} else {
@@ -109,7 +111,7 @@ public class DatadogHttpReporter implements MetricReporter, Scheduled {
 		Integer proxyPort = config.getInteger(PROXY_PORT, 8080);
 		String tags = config.getString(TAGS, "");
 
-		client = new DatadogHttpClient(apiKey, proxyHost, proxyPort);
+		client = new DatadogHttpClient(apiKey, proxyHost, proxyPort, true);
 
 		configTags = getTagsFromConfig(tags);
 

flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java

@@ -24,14 +24,7 @@ import org.apache.flink.metrics.Meter;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 
-import org.junit.Before;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.api.support.membermodification.MemberMatcher;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
 
 import java.net.InetSocketAddress;
 import java.net.Proxy;
@@ -44,45 +37,31 @@ import static org.junit.Assert.assertTrue;
 /**
  * Tests for the DatadogHttpClient.
  */
-@RunWith(PowerMockRunner.class)
-@PrepareForTest({DMetric.class, DatadogHttpClient.class})
-@PowerMockIgnore({"javax.net.ssl.*", "javax.management.*"})
 public class DatadogHttpClientTest {
 
 	private static List<String> tags = Arrays.asList("tag1", "tag2");
 
 	private static final long MOCKED_SYSTEM_MILLIS = 123L;
 
-	@Before
-	public void mockSystemMillis() {
-		PowerMockito.mockStatic(DMetric.class);
-		PowerMockito.when(DMetric.getUnixEpochTimestamp()).thenReturn(MOCKED_SYSTEM_MILLIS);
-	}
-
-	@Before
-	public void suppressValidateApiKey() {
-		PowerMockito.suppress(MemberMatcher.method(DatadogHttpClient.class, "validateApiKey"));
-	}
-
 	@Test(expected = IllegalArgumentException.class)
 	public void testClientWithEmptyKey() {
-		new DatadogHttpClient("", null, 123);
+		new DatadogHttpClient("", null, 123, false);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testClientWithNullKey() {
-		new DatadogHttpClient(null, null, 123);
+		new DatadogHttpClient(null, null, 123, false);
 	}
 
 	@Test
 	public void testGetProxyWithNullProxyHost() {
-		DatadogHttpClient client = new DatadogHttpClient("anApiKey", null, 123);
+		DatadogHttpClient client = new DatadogHttpClient("anApiKey", null, 123, false);
 		assert(client.getProxy() == Proxy.NO_PROXY);
 	}
 
 	@Test
 	public void testGetProxy() {
-		DatadogHttpClient client = new DatadogHttpClient("anApiKey", "localhost", 123);
+		DatadogHttpClient client = new DatadogHttpClient("anApiKey", "localhost", 123, false);
 
 		assertTrue(client.getProxy().address() instanceof InetSocketAddress);
 
@@ -100,7 +79,7 @@ public class DatadogHttpClientTest {
 			public Number getValue() {
 				return 1;
 			}
-		}, "testCounter", "localhost", tags);
+		}, "testCounter", "localhost", tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testCounter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}",
@@ -115,7 +94,7 @@ public class DatadogHttpClientTest {
 			public Number getValue() {
 				return 1;
 			}
-		}, "testCounter", null, tags);
+		}, "testCounter", null, tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testCounter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}",
@@ -141,7 +120,7 @@ public class DatadogHttpClientTest {
 			public long getCount() {
 				return 1;
 			}
-		}, "testCounter", "localhost", tags);
+		}, "testCounter", "localhost", tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testCounter\",\"type\":\"count\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}",
@@ -167,7 +146,7 @@ public class DatadogHttpClientTest {
 			public long getCount() {
 				return 1;
 			}
-		}, "testCounter", null, tags);
+		}, "testCounter", null, tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testCounter\",\"type\":\"count\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}",
@@ -193,7 +172,7 @@ public class DatadogHttpClientTest {
 			public long getCount() {
 				return 0;
 			}
-		}, "testMeter", "localhost", tags);
+		}, "testMeter", "localhost", tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testMeter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}",
@@ -219,7 +198,7 @@ public class DatadogHttpClientTest {
 			public long getCount() {
 				return 0;
 			}
-		}, "testMeter", null, tags);
+		}, "testMeter", null, tags, () -> MOCKED_SYSTEM_MILLIS);
 
 		assertEquals(
 			"{\"metric\":\"testMeter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}",