@Test
public void testCollectMetrics() {

  // Given
  TimeUtils time = new TimeUtils();
  Uuid uuid = new Uuid();
  long unixTimeAtTestStart = time.nowInUnixTime();
  Collector metricsCollector = new BasicCollector(mockServer, time, uuid);
  // When
  GenericContainer metricsRecord = metricsCollector.collectMetrics();
  // Then
  assertThat(metricsRecord).isInstanceOf(SupportKafkaMetricsBasic.class);
  assertThat(metricsRecord.getSchema()).isEqualTo(SupportKafkaMetricsBasic.getClassSchema());
  SupportKafkaMetricsBasic basicRecord = (SupportKafkaMetricsBasic) metricsRecord;
  assertThat(basicRecord.getTimestamp()).isBetween(unixTimeAtTestStart, time.nowInUnixTime());
  assertThat(basicRecord.getKafkaVersion()).isEqualTo(AppInfoParser.getVersion());
  assertThat(basicRecord.getConfluentPlatformVersion()).isEqualTo(Version.getVersion());
  assertThat(basicRecord.getCollectorState()).isEqualTo(metricsCollector.getRuntimeState().stateId());
  assertThat(basicRecord.getBrokerProcessUUID()).isEqualTo(uuid.toString());
}