private void maybeUpdateGroupSubscription(String assignorName,
                                              Map<String, Assignment> assignments,
                                              Set<String> allSubscribedTopics) {
        if (!isAssignFromSubscribedTopicsAssignor(assignorName)) {
            Set<String> assignedTopics = new HashSet<>();
            for (Assignment assigned : assignments.values()) {
                for (TopicPartition tp : assigned.partitions())
                    assignedTopics.add(tp.topic());
            }

            if (!assignedTopics.containsAll(allSubscribedTopics)) {
                SortedSet<String> notAssignedTopics = new TreeSet<>(allSubscribedTopics);
                notAssignedTopics.removeAll(assignedTopics);
                log.warn("The following subscribed topics are not assigned to any members: {} ", notAssignedTopics);
            }

            if (!allSubscribedTopics.containsAll(assignedTopics)) {
                SortedSet<String> newlyAddedTopics = new TreeSet<>(assignedTopics);
                newlyAddedTopics.removeAll(allSubscribedTopics);
                log.info("The following not-subscribed topics are assigned, and their metadata will be " +
                    "fetched from the brokers: {}", newlyAddedTopics);

                allSubscribedTopics.addAll(newlyAddedTopics);
                updateGroupSubscription(allSubscribedTopics);
            }
        }
    }
user-customized assignor may have created some topics that are not in the subscription list
and assign their partitions to the members; in this case we would like to update the leader's
own metadata with the newly added topics so that it will not trigger a subsequent rebalance
when these topics gets updated from metadata refresh.

We skip the check for in-product assignors since this will not happen in in-product assignors.

TODO: this is a hack and not something we want to support long-term unless we push regex into the protocol
      we may need to modify the ConsumerPartitionAssignor API to better support this case.

@param assignorName          the selected assignor name
@param assignments           the assignments after assignor assigned
@param allSubscribedTopics   all consumers' subscribed topics
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java