### Question:
MonitorOffsetsConsumerInterceptor implements ConsumerInterceptor<K, V> { @Override public ConsumerRecords<K, V> onConsume(final ConsumerRecords<K, V> records) { for (TopicPartition tp : records.partitions()) { for (ConsumerRecord<K, V> record : records.records(tp)) { final OffsetAndTimestamp ot = new OffsetAndTimestamp(record.offset(), record.timestamp()); consumerGroupOffsets.update(tp, consumerThreadKey(), current -> current.consumedOffset(ot)); } } return records; } MonitorOffsetsConsumerInterceptor(); @Override void configure(final Map<String, ?> configs); @Override ConsumerRecords<K, V> onConsume(final ConsumerRecords<K, V> records); @Override void onCommit(final Map<TopicPartition, OffsetAndMetadata> offsets); @Override void close();  }### Answer:
@Test public void shouldUpdateConsumerGroupStateWhenCallingOnConsume() { final ConsumerRecord<String, String> record = new ConsumerRecord<>( TOPIC_PARTITION.topic(), TOPIC_PARTITION.partition(), 123L, "k", "v" ); interceptor.onConsume(new ConsumerRecords<>(Collections.singletonMap( TOPIC_PARTITION, Collections.singletonList(record)) ) ); final ConsumerGroupOffsetsState state = GlobalConsumerOffsetsRegistry.getInstance().offsetsFor(TEST_GROUP); final ConsumerLogOffsets logOffsets = state.offsets().get(TOPIC_PARTITION).right(); Assertions.assertEquals(record.offset(), logOffsets.consumedOffset().offset()); Assertions.assertEquals(record.timestamp(), logOffsets.consumedOffset().timestamp()); }