@Override
  public KsqlVersionMetrics collectMetrics() {
    final KsqlVersionMetrics metricsRecord = new KsqlVersionMetrics();
    metricsRecord.setTimestamp(TimeUnit.MILLISECONDS.toSeconds(clock.millis()));
    metricsRecord.setConfluentPlatformVersion(AppInfo.getVersion());
    metricsRecord.setKsqlComponentType(moduleType.name());
    metricsRecord.setIsActive(activenessSupplier.get());
    return metricsRecord;
  }
@Test
  public void testCollectMetricsAssignsCurrentTime() {
    // Given:
    final long t1 = 1000L;
    final long t2 = 1001L;
    when(clock.millis())
        .thenReturn(TimeUnit.SECONDS.toMillis(t1))
        .thenReturn(TimeUnit.SECONDS.toMillis(t2));

    // When:
    KsqlVersionMetrics metrics = basicCollector.collectMetrics();

    // Then:
    assertThat(metrics.getTimestamp(), is(t1));

    metrics = basicCollector.collectMetrics();
    assertThat(metrics.getTimestamp(), is(t2));
  }