docs: @Override
    public Map<String, Assignment> assign(final Cluster metadata,
                                          final Map<String, Subscription> subscriptions) {
      // construct the client metadata from the decoded subscription info
      final Map<UUID, ClientMetadata> clientsMetadata = new HashMap<>();
      final Set<String> futureConsumers = new HashSet<>();
  
      int minReceivedMetadataVersion = SubscriptionInfo.LATEST_SUPPORTED_VERSION;
  
      supportedVersions.clear();
      int futureMetadataVersion = UNKNOWN;
      for (final Map.Entry<String, Subscription> entry : subscriptions.entrySet()) {
        final String consumerId = entry.getKey();
        final Subscription subscription = entry.getValue();
  
        final SubscriptionInfo info = SubscriptionInfo.decode(subscription.userData());
        final int usedVersion = info.version();
        supportedVersions.add(info.latestSupportedVersion());
        if (usedVersion > SubscriptionInfo.LATEST_SUPPORTED_VERSION) {
          futureMetadataVersion = usedVersion;
          futureConsumers.add(consumerId);
          continue;
        }
        if (usedVersion < minReceivedMetadataVersion) {
          minReceivedMetadataVersion = usedVersion;
        }
  
        // create the new client metadata if necessary
        ClientMetadata clientMetadata = clientsMetadata.get(info.processId());
  
        if (clientMetadata == null) {
          clientMetadata = new ClientMetadata(info.userEndPoint());
          clientsMetadata.put(info.processId(), clientMetadata);
        }
  
        // add the consumer to the client
        clientMetadata.addConsumer(consumerId, info);
      }
  
      final boolean versionProbing;
      if (futureMetadataVersion != UNKNOWN) {
        if (minReceivedMetadataVersion >= EARLIEST_PROBEABLE_VERSION) {
          log.info("Received a future (version probing) subscription (version: {}). Sending empty assignment back (with supported version {}).",
              futureMetadataVersion,
              SubscriptionInfo.LATEST_SUPPORTED_VERSION);
          versionProbing = true;
        } else {
          throw new IllegalStateException("Received a future (version probing) subscription (version: " + futureMetadataVersion
              + ") and an incompatible pre Kafka 2.0 subscription (version: " + minReceivedMetadataVersion + ") at the same time.");
        }
      } else {
        versionProbing = false;
      }
  
      if (minReceivedMetadataVersion < SubscriptionInfo.LATEST_SUPPORTED_VERSION) {
        log.info("Downgrading metadata to version {}. Latest supported version is {}.",
            minReceivedMetadataVersion,
            SubscriptionInfo.LATEST_SUPPORTED_VERSION);
      }
  
      log.debug("Constructed client metadata {} from the member subscriptions.", clientsMetadata);
  
      // ---------------- Step Zero ---------------- //
  
      // parse the topology to determine the repartition source topics,
      // making sure they are created with the number of partitions as
      // the maximum of the depending sub-topologies source topics' number of partitions
      final Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = taskManager.builder().topicGroups();
  
      final Map<String, InternalTopicMetadata> repartitionTopicMetadata = new HashMap<>();
      for (final InternalTopologyBuilder.TopicsInfo topicsInfo : topicGroups.values()) {
        for (final String topic : topicsInfo.sourceTopics) {
          if (!topicsInfo.repartitionSourceTopics.keySet().contains(topic) &&
              !metadata.topics().contains(topic)) {
            return errorAssignment(clientsMetadata, topic, Error.INCOMPLETE_SOURCE_TOPIC_METADATA.code);
          }
        }
        for (final InternalTopicConfig topic: topicsInfo.repartitionSourceTopics.values()) {
          repartitionTopicMetadata.put(topic.name(), new InternalTopicMetadata(topic));
        }
      }
  
      boolean numPartitionsNeeded;
      do {
        numPartitionsNeeded = false;
  
        for (final InternalTopologyBuilder.TopicsInfo topicsInfo : topicGroups.values()) {
          for (final String topicName : topicsInfo.repartitionSourceTopics.keySet()) {
            int numPartitions = repartitionTopicMetadata.get(topicName).numPartitions;
  
            // try set the number of partitions for this repartition topic if it is not set yet
            if (numPartitions == UNKNOWN) {
              for (final InternalTopologyBuilder.TopicsInfo otherTopicsInfo : topicGroups.values()) {
                final Set<String> otherSinkTopics = otherTopicsInfo.sinkTopics;
  
                if (otherSinkTopics.contains(topicName)) {
                  // if this topic is one of the sink topics of this topology,
                  // use the maximum of all its source topic partitions as the number of partitions
                  for (final String sourceTopicName : otherTopicsInfo.sourceTopics) {
                    final Integer numPartitionsCandidate;
                    // It is possible the sourceTopic is another internal topic, i.e,
                    // map().join().join(map())
                    if (repartitionTopicMetadata.containsKey(sourceTopicName)) {
                      numPartitionsCandidate = repartitionTopicMetadata.get(sourceTopicName).numPartitions;
                    } else {
                      numPartitionsCandidate = metadata.partitionCountForTopic(sourceTopicName);
                    }
  
                    if (numPartitionsCandidate > numPartitions) {
                      numPartitions = numPartitionsCandidate;
                    }
                  }
                }
              }
              // if we still have not find the right number of partitions,
              // another iteration is needed
              if (numPartitions == UNKNOWN) {
                numPartitionsNeeded = true;
              } else {
                repartitionTopicMetadata.get(topicName).numPartitions = numPartitions;
              }
            }
          }
        }
      } while (numPartitionsNeeded);
  
  
      // ensure the co-partitioning topics within the group have the same number of partitions,
      // and enforce the number of partitions for those repartition topics to be the same if they
      // are co-partitioned as well.
      ensureCopartitioning(taskManager.builder().copartitionGroups(), repartitionTopicMetadata, metadata);
  
      // make sure the repartition source topics exist with the right number of partitions,
      // create these topics if necessary
      prepareTopic(repartitionTopicMetadata);
  
      // augment the metadata with the newly computed number of partitions for all the
      // repartition source topics
      final Map<TopicPartition, PartitionInfo> allRepartitionTopicPartitions = new HashMap<>();
      for (final Map.Entry<String, InternalTopicMetadata> entry : repartitionTopicMetadata.entrySet()) {
        final String topic = entry.getKey();
        final int numPartitions = entry.getValue().numPartitions;
  
        for (int partition = 0; partition < numPartitions; partition++) {
          allRepartitionTopicPartitions.put(new TopicPartition(topic, partition),
              new PartitionInfo(topic, partition, null, new Node[0], new Node[0]));
        }
      }
  
      final Cluster fullMetadata = metadata.withPartitions(allRepartitionTopicPartitions);
      taskManager.setClusterMetadata(fullMetadata);
  
      log.debug("Created repartition topics {} from the parsed topology.", allRepartitionTopicPartitions.values());
  
      // ---------------- Step One ---------------- //
  
      // get the tasks as partition groups from the partition grouper
      final Set<String> allSourceTopics = new HashSet<>();
      final Map<Integer, Set<String>> sourceTopicsByGroup = new HashMap<>();
      for (final Map.Entry<Integer, InternalTopologyBuilder.TopicsInfo> entry : topicGroups.entrySet()) {
        allSourceTopics.addAll(entry.getValue().sourceTopics);
        sourceTopicsByGroup.put(entry.getKey(), entry.getValue().sourceTopics);
      }
  
      final Map<TaskId, Set<TopicPartition>> partitionsForTask = partitionGrouper.partitionGroups(sourceTopicsByGroup, fullMetadata);
  
      // check if all partitions are assigned, and there are no duplicates of partitions in multiple tasks
      final Set<TopicPartition> allAssignedPartitions = new HashSet<>();
      final Map<Integer, Set<TaskId>> tasksByTopicGroup = new HashMap<>();
      for (final Map.Entry<TaskId, Set<TopicPartition>> entry : partitionsForTask.entrySet()) {
        final Set<TopicPartition> partitions = entry.getValue();
        for (final TopicPartition partition : partitions) {
          if (allAssignedPartitions.contains(partition)) {
            log.warn("Partition {} is assigned to more than one tasks: {}", partition, partitionsForTask);
          }
        }
        allAssignedPartitions.addAll(partitions);
  
        final TaskId id = entry.getKey();
        tasksByTopicGroup.computeIfAbsent(id.topicGroupId, k -> new HashSet<>()).add(id);
      }
      for (final String topic : allSourceTopics) {
        final List<PartitionInfo> partitionInfoList = fullMetadata.partitionsForTopic(topic);
        if (!partitionInfoList.isEmpty()) {
          for (final PartitionInfo partitionInfo : partitionInfoList) {
            final TopicPartition partition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
            if (!allAssignedPartitions.contains(partition)) {
              log.warn("Partition {} is not assigned to any tasks: {}"
                  + " Possible causes of a partition not getting assigned"
                  + " is that another topic defined in the topology has not been"
                  + " created when starting your streams application,"
                  + " resulting in no tasks created for this topology at all.", partition, partitionsForTask);
            }
          }
        } else {
          log.warn("No partitions found for topic {}", topic);
        }
      }
  
      // add tasks to state change log topic subscribers
      final Map<String, InternalTopicMetadata> changelogTopicMetadata = new HashMap<>();
      for (final Map.Entry<Integer, InternalTopologyBuilder.TopicsInfo> entry : topicGroups.entrySet()) {
        final int topicGroupId = entry.getKey();
        final Map<String, InternalTopicConfig> stateChangelogTopics = entry.getValue().stateChangelogTopics;
  
        for (final InternalTopicConfig topicConfig : stateChangelogTopics.values()) {
          // the expected number of partitions is the max value of TaskId.partition + 1
          int numPartitions = UNKNOWN;
          if (tasksByTopicGroup.get(topicGroupId) != null) {
            for (final TaskId task : tasksByTopicGroup.get(topicGroupId)) {
              if (numPartitions < task.partition + 1)
                numPartitions = task.partition + 1;
            }
            final InternalTopicMetadata topicMetadata = new InternalTopicMetadata(topicConfig);
            topicMetadata.numPartitions = numPartitions;
  
            changelogTopicMetadata.put(topicConfig.name(), topicMetadata);
          } else {
            log.debug("No tasks found for topic group {}", topicGroupId);
          }
        }
      }
  
      prepareTopic(changelogTopicMetadata);
  
      log.debug("Created state changelog topics {} from the parsed topology.", changelogTopicMetadata.values());
  
      // ---------------- Step Two ---------------- //
  
      // assign tasks to clients
      final Map<UUID, ClientState> states = new HashMap<>();
      for (final Map.Entry<UUID, ClientMetadata> entry : clientsMetadata.entrySet()) {
        states.put(entry.getKey(), entry.getValue().state);
      }
  
      log.debug("Assigning tasks {} to clients {} with number of replicas {}",
          partitionsForTask.keySet(), states, numStandbyReplicas);
  
      //TWITTER CHANGED
      //final StickyTaskAssignor<UUID> taskAssignor = new StickyTaskAssignor<>(states, partitionsForTask.keySet());
      final TaskAssignor<UUID, TaskId> taskAssignor = createTaskAssignor(partitionsForTask, states, clientsMetadata);
      taskAssignor.assign(numStandbyReplicas);
  
      log.info("Assigned tasks to clients as {}.", states);
  
      // ---------------- Step Three ---------------- //
  
      // construct the global partition assignment per host map
      final Map<HostInfo, Set<TopicPartition>> partitionsByHostState = new HashMap<>();
      if (minReceivedMetadataVersion >= 2) {
        for (final Map.Entry<UUID, ClientMetadata> entry : clientsMetadata.entrySet()) {
          final HostInfo hostInfo = entry.getValue().hostInfo;
  
          if (hostInfo != null) {
            final Set<TopicPartition> topicPartitions = new HashSet<>();
            final ClientState state = entry.getValue().state;
  
            for (final TaskId id : state.activeTasks()) {
              topicPartitions.addAll(partitionsForTask.get(id));
            }
  
            partitionsByHostState.put(hostInfo, topicPartitions);
          }
        }
      }
      taskManager.setPartitionsByHostState(partitionsByHostState);
  
      final Map<String, Assignment> assignment;
      if (versionProbing) {
        assignment = versionProbingAssignment(clientsMetadata, partitionsForTask, partitionsByHostState, futureConsumers, minReceivedMetadataVersion);
      } else {
        assignment = computeNewAssignment(clientsMetadata, partitionsForTask, partitionsByHostState, minReceivedMetadataVersion);
      }
  
      return assignment;
    }