### Question:
FreshnessTracker implements Runnable { public void run() { long freshness = 0; if (!this.consumer.upToDate) { long firstUncommittedOffset = consumer.offset + 1; Optional<Long> entered = this.getOffsetIngestTime(firstUncommittedOffset); metrics.kafkaRead.labels(this.consumer.cluster, this.consumer.group).inc(); if (!entered.isPresent()) { LOG.error("Failed to load offset for {} ({}) : {}", consumer.group, consumer.tp, firstUncommittedOffset); metrics.failed.labels(this.consumer.cluster, this.consumer.group).inc(); return; } if (entered.get() < 0) { LOG.error("Read invalid offset for {} ({}) : {}", consumer.group, consumer.tp, entered.get()); metrics.invalid.labels(this.consumer.cluster, this.consumer.group).inc(); return; } Instant now = Instant.now(clock); freshness = Math.max(now.toEpochMilli() - entered.get(), 0); } metrics.freshness.labels(this.consumer.cluster, this.consumer.group, this.from.topic(), Integer.toString(this.from.partition())) .set(freshness); } FreshnessTracker(ConsumerOffset consumer, Consumer kafka, FreshnessMetrics metrics); void run();  }### Answer:
@Test public void testNoLag() throws Exception { FreshnessTracker.ConsumerOffset consumer = new FreshnessTracker.ConsumerOffset("cluster", "group", "topic", 1, 1, true); FreshnessMetrics metrics = new FreshnessMetrics(); Consumer kafka = mock(Consumer.class); FreshnessTracker work = new FreshnessTracker(consumer, kafka, metrics); work.run(); assertEquals("Freshness should be 0 if no lag", 0, metrics.freshness.labels("cluster", "group", "topic", "1").get(), 0); assertEquals(0, metrics.failed.labels("cluster", "group").get(), 0); verifyZeroInteractions(kafka); }
@Test public void testInvalidOffset() { FreshnessTracker.ConsumerOffset consumer = new FreshnessTracker.ConsumerOffset("cluster", "group", "topic", 1, 1, false); ConsumerRecords records = createConsumerRecordsAtTimestamps(-1L); Consumer kafka = mock(Consumer.class); when(kafka.poll(any(Duration.class))).thenReturn(records); FreshnessMetrics metrics = new FreshnessMetrics(); new FreshnessTracker(consumer, kafka, metrics).run(); assertEquals(1, metrics.invalid.labels("cluster", "group").get(), 0); assertEquals(0, metrics.freshness.labels("cluster", "group", "topic", "1").get(), 0); }
@Test public void testCouldNotLoadLatestTimestamp() throws Exception { ConsumerRecords logEndOffsetTimestamp = createConsumerRecordsAtTimestamps(); Consumer kafka = mock(Consumer.class); when(kafka.poll(Duration.ofSeconds(2))).thenReturn(logEndOffsetTimestamp); FreshnessTracker.ConsumerOffset consumer = new FreshnessTracker.ConsumerOffset("cluster", "group", "topic", 1, 1, false); FreshnessMetrics metrics = new FreshnessMetrics(); FreshnessTracker work = new FreshnessTracker(consumer, kafka, metrics); work.run(); assertEquals("Freshness should not be set", 0, metrics.freshness.labels("cluster", "group", "topic", "1").get(), 0); verify(kafka).poll(Duration.ofSeconds(2)); TopicPartition tp = new TopicPartition("topic", 1); Collection<TopicPartition> tps = Collections.singletonList(tp); verify(kafka).assign(tps); verify(kafka).seek(tp, 1); verifyNoMoreInteractions(kafka); }