class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Configurable {

    @Override
    public void configure(final Map<String, ?> configs) {
        final AssignorConfiguration assignorConfiguration = new AssignorConfiguration(configs);

        logPrefix = assignorConfiguration.logPrefix();
        log = new LogContext(logPrefix).logger(getClass());
        usedSubscriptionMetadataVersion = assignorConfiguration
            .configuredMetadataVersion(usedSubscriptionMetadataVersion);
        taskManager = assignorConfiguration.taskManager();
        streamsMetadataState = assignorConfiguration.streamsMetadataState();
        assignmentErrorCode = assignorConfiguration.assignmentErrorCode();
        nextScheduledRebalanceMs = assignorConfiguration.nextScheduledRebalanceMs();
        time = assignorConfiguration.time();
        assignmentConfigs = assignorConfiguration.assignmentConfigs();
        partitionGrouper = assignorConfiguration.partitionGrouper();
        userEndPoint = assignorConfiguration.userEndPoint();
        adminClient = assignorConfiguration.adminClient();
        internalTopicManager = assignorConfiguration.internalTopicManager();
        copartitionedTopicsEnforcer = assignorConfiguration.copartitionedTopicsEnforcer();
        rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
        taskAssignorSupplier = assignorConfiguration::taskAssignor;
        assignmentListener = assignorConfiguration.assignmentListener();
    }

}