class StreamPartitionAssignor implements PartitionAssignor, Configurable {

    @Override
    public Subscription subscription(Set<String> topics) {
        // Adds the following information to subscription
        // 1. Client UUID (a unique id assigned to an instance of KafkaStreams)
        // 2. Task ids of previously running tasks
        // 3. Task ids of valid local states on the client's state directory.

        final Set<TaskId> previousActiveTasks = taskManager.prevActiveTaskIds();
        final Set<TaskId> standbyTasks = taskManager.cachedTasksIds();
        standbyTasks.removeAll(previousActiveTasks);
        final SubscriptionInfo data = new SubscriptionInfo(taskManager.processId(), previousActiveTasks, standbyTasks, this.userEndPoint);

        taskManager.updateSubscriptionsFromMetadata(topics);

        return new Subscription(new ArrayList<>(topics), data.encode());
    }

}