@Override
    public Map<TopicPartition, Long> offsets() {
        return Collections.unmodifiableMap(new HashMap<>(offsets));
    }
@throws StreamsException fatal error that should cause the thread to die
@throws TaskMigratedException recoverable error that would cause the task to be removed
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java