@Override
    void updateLags() {
        if (maxTaskIdleMs != StreamsConfig.MAX_TASK_IDLE_MS_DISABLED) {
            for (final TopicPartition tp : partitionQueues.keySet()) {
                final OptionalLong l = lagProvider.apply(tp);
                if (l.isPresent()) {
                    fetchedLags.put(tp, l.getAsLong());
                    logger.trace("Updated lag for {} to {}", tp, l.getAsLong());
                } else {
                    fetchedLags.remove(tp);
                }
            }
        }
    }
@throws IllegalStateException if the record's partition does not belong to this partition group
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java