### Question:
MonitorOffsetsConsumerInterceptor implements ConsumerInterceptor<K, V> { @Override public void onCommit(final Map<TopicPartition, OffsetAndMetadata> offsets) { final long now = Time.SYSTEM.milliseconds(); for (Map.Entry<TopicPartition, OffsetAndMetadata> elem : offsets.entrySet()) { final TopicPartition tp = elem.getKey(); final OffsetAndMetadata oam = elem.getValue(); final OffsetAndTimestamp ot = new OffsetAndTimestamp(oam.offset(), now); consumerGroupOffsets.update(tp, consumerThreadKey(), current -> current.committedOffset(ot)); } } 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 shouldUpdateConsumerGroupStateWhenCallingOnCommit() { interceptor.onCommit(Collections.singletonMap(TOPIC_PARTITION, new OffsetAndMetadata(123L))); final ConsumerGroupOffsetsState state = GlobalConsumerOffsetsRegistry.getInstance().offsetsFor(TEST_GROUP); final ConsumerLogOffsets logOffsets = state.offsets().get(TOPIC_PARTITION).right(); Assertions.assertEquals(123L, logOffsets.committedOffset().offset()); }