public long lagFor(final TaskId task) {
        final Long totalLag = taskLagTotals.get(task);
        if (totalLag == null) {
            throw new IllegalStateException("Tried to lookup lag for unknown task " + task);
        }
        return totalLag;
    }
Returns the total lag across all logged stores in the task. Equal to the end offset sum if this client
did not have any state for this task on disk.

@return end offset sum - offset sum
         Task.LATEST_OFFSET if this was previously an active running task on this client
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java