@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);
                }
            }
        }
    }
@Test
    public void shouldUpdateLags() {
        final HashMap<TopicPartition, OptionalLong> lags = new HashMap<>();
        final PartitionGroup group = new PartitionGroup(
            logContext,
            mkMap(
                mkEntry(partition1, queue1)
            ),
            tp -> lags.getOrDefault(tp, OptionalLong.empty()),
            getValueSensor(metrics, lastLatenessValue),
            enforcedProcessingSensor,
            10L
        );

        hasNoFetchedLag(group, partition1);

        lags.put(partition1, OptionalLong.of(5));

        hasNoFetchedLag(group, partition1);

        group.updateLags();

        hasNonZeroFetchedLag(group, partition1, 5);

        lags.put(partition1, OptionalLong.of(0));
        group.updateLags();

        hasZeroFetchedLag(group, partition1);

        lags.remove(partition1);

        hasZeroFetchedLag(group, partition1);

        group.updateLags();

        hasNoFetchedLag(group, partition1);
    }