class ConsumerCoordinator extends AbstractCoordinator {

    @Override
    public boolean rejoinNeededOrPending() {
        if (!subscriptions.hasAutoAssignedPartitions())
            return false;

        // we need to rejoin if we performed the assignment and metadata has changed;
        // also for those owned-but-no-longer-existed partitions we should drop them as lost
        if (assignmentSnapshot != null && !assignmentSnapshot.matches(metadataSnapshot)) {
            log.info("Requesting to re-join the group and trigger rebalance since the assignment metadata has changed from {} to {}",
                    assignmentSnapshot, metadataSnapshot);

            requestRejoin();
            return true;
        }

        // we need to join if our subscription has changed since the last join
        if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subscription())) {
            log.info("Requesting to re-join the group and trigger rebalance since the subscription has changed from {} to {}",
                joinedSubscription, subscriptions.subscription());

            requestRejoin();
            return true;
        }

        return super.rejoinNeededOrPending();
    }

    public  ConsumerCoordinator(GroupRebalanceConfig rebalanceConfig,
                               LogContext logContext,
                               ConsumerNetworkClient client,
                               List<ConsumerPartitionAssignor> assignors,
                               ConsumerMetadata metadata,
                               SubscriptionState subscriptions,
                               Metrics metrics,
                               String metricGrpPrefix,
                               Time time,
                               boolean autoCommitEnabled,
                               int autoCommitIntervalMs,
                               ConsumerInterceptors<?, ?> interceptors,
                               boolean throwOnFetchStableOffsetsUnsupported);

}