@Override
        public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, Subscription> subscriptions) {
            subscriptions.forEach((consumer, subscription) -> {
                if (!subscription.rackId().isPresent())
                    throw new IllegalStateException("Rack id not provided in subscription for " + consumer);
                rackIds.add(subscription.rackId().get());
            });
            return super.assign(partitionsPerTopic, subscriptions);
        }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java