private InternalTopologyBuilder lookupBuilderForTask(final TaskId task) {
        final InternalTopologyBuilder builder = task.topologyName() == null ?
            builders.get(UNNAMED_TOPOLOGY) :
            builders.get(task.topologyName());
        if (builder == null) {
            throw new UnknownTopologyException("Unable to locate topology builder", task.topologyName());
        } else {
            return builder;
        }
    }
@return the {@link InternalTopologyBuilder} for this task's topology, guaranteed to be non-null

@throws UnknownTopologyException  if the task is from a named topology that this client isn't aware of
streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java