private Map<String, Assignment> errorAssignment(final Map<ProcessId, ClientMetadata> clientsMetadata,
                                                    final int errorCode) {
        final Map<String, Assignment> assignment = new HashMap<>();
        for (final ClientMetadata clientMetadata : clientsMetadata.values()) {
            for (final String consumerId : clientMetadata.consumers) {
                assignment.put(consumerId, new Assignment(
                    Collections.emptyList(),
                    new AssignmentInfo(LATEST_SUPPORTED_VERSION,
                        Collections.emptyList(),
                        Collections.emptyMap(),
                        Collections.emptyMap(),
                        Collections.emptyMap(),
                        errorCode).encode()
                ));
            }
        }
        return assignment;
    }
We need to have the PartitionAssignor and its StreamThread to be mutually accessible since the former needs
latter's cached metadata while sending subscriptions, and the latter needs former's returned assignment when
adding tasks.

@throws KafkaException if the stream thread is not specified
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java