@Override
    void clear() {
        for (final RecordQueue queue : partitionQueues.values()) {
            queue.clear();
        }
        nonEmptyQueuesByTime.clear();
        totalBuffered = 0;
        streamTime = RecordQueue.UNKNOWN;
        fetchedLags.clear();
    }
@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