Make sure each Beacon ping includes current time (#681)

ksql-version-metrics-client/src/main/java/io/confluent/ksql/version/metrics/collector/BasicCollector.java

@@ -25,18 +25,18 @@
 
 public class BasicCollector extends Collector {
 
-  private final long time;
+  private final TimeUtils timeUtils;
   private final KsqlModuleType moduleType;
 
   public BasicCollector(KsqlModuleType moduleType, TimeUtils timeUtils) {
-    time = timeUtils.nowInUnixTime();
+    this.timeUtils = timeUtils;
     this.moduleType = moduleType;
   }
 
   @Override
   public GenericContainer collectMetrics() {
     KsqlVersionMetrics metricsRecord = new KsqlVersionMetrics();
-    metricsRecord.setTimestamp(time);
+    metricsRecord.setTimestamp(timeUtils.nowInUnixTime());
     metricsRecord.setConfluentPlatformVersion(Version.getVersion());
     metricsRecord.setKsqlComponentType(moduleType.name());
     return metricsRecord;

ksql-version-metrics-client/src/test/java/io/confluent/ksql/version/metrics/collector/BasicCollectorTest.java

@@ -16,23 +16,41 @@
 
 package io.confluent.ksql.version.metrics.collector;
 
-import org.easymock.EasyMock;
 import org.hamcrest.CoreMatchers;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.Before;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import static org.junit.Assert.assertEquals;
+
 import java.util.Collection;
 import java.util.EnumSet;
 
 import io.confluent.ksql.version.metrics.KsqlVersionMetrics;
 import io.confluent.support.metrics.common.Version;
+import io.confluent.support.metrics.common.time.Clock;
 import io.confluent.support.metrics.common.time.TimeUtils;
 
 @RunWith(Parameterized.class)
 public class BasicCollectorTest {
 
+  public class MockClock implements Clock {
+    private long currentTime = 0;
+
+    public MockClock() {
+    }
+
+    public long currentTimeMs() {
+      return currentTime;
+    }
+
+    public void setCurrentTimeMillis(long timeMillis) {
+      currentTime = timeMillis;
+    }
+  }
+
   @Parameterized.Parameters
   public static Collection<KsqlModuleType> data() {
     return EnumSet.allOf(KsqlModuleType.class);
@@ -41,20 +59,44 @@ public static Collection<KsqlModuleType> data() {
   @Parameterized.Parameter
   public KsqlModuleType moduleType;
 
-  @Test
-  public void testGetCollector(){
+  private MockClock mockClock;
+  private TimeUtils timeUtils;
+
+  @Before
+  public void setUp() throws Exception {
+    mockClock = new MockClock();
+    timeUtils = new TimeUtils(mockClock);
+  }
 
-    TimeUtils timeUtils = EasyMock.mock(TimeUtils.class);
-    EasyMock.expect(timeUtils.nowInUnixTime()).andReturn(System.currentTimeMillis()).anyTimes();
-    EasyMock.replay(timeUtils);
+  @Test
+  public void testGetCollector() {
     BasicCollector basicCollector = new BasicCollector(moduleType, timeUtils);
 
     KsqlVersionMetrics expectedMetrics = new KsqlVersionMetrics();
     expectedMetrics.setTimestamp(timeUtils.nowInUnixTime());
     expectedMetrics.setConfluentPlatformVersion(Version.getVersion());
     expectedMetrics.setKsqlComponentType(moduleType.name());
 
+    // should match because we don't advance the clock
     Assert.assertThat(basicCollector.collectMetrics(), CoreMatchers.equalTo(expectedMetrics));
   }
 
+  @Test
+  public void testCollectMetricsAssignsCurrentTime() {
+    Long currentTimeSec = 1000l;
+
+    mockClock.setCurrentTimeMillis(currentTimeSec * 1000);
+    BasicCollector basicCollector = new BasicCollector(moduleType, timeUtils);
+
+    currentTimeSec += 12300l;
+    mockClock.setCurrentTimeMillis(currentTimeSec * 1000);
+    KsqlVersionMetrics metrics = (KsqlVersionMetrics) basicCollector.collectMetrics();
+    assertEquals(currentTimeSec, metrics.getTimestamp());
+
+    currentTimeSec += 734l;
+    mockClock.setCurrentTimeMillis(currentTimeSec * 1000);
+    metrics = (KsqlVersionMetrics) basicCollector.collectMetrics();
+    assertEquals(currentTimeSec, metrics.getTimestamp());
+  }
+
 }