docs: @Override
    public void configure(final Map<String, ?> configs) {
      final StreamsConfig streamsConfig = new InternalStreamsConfig(configs);
  
      // Setting the logger with the passed in client thread name
      logPrefix = String.format("stream-thread [%s] ", streamsConfig.getString(CommonClientConfigs.CLIENT_ID_CONFIG));
      final LogContext logContext = new LogContext(logPrefix);
      log = logContext.logger(getClass());
  
      final String upgradeFrom = streamsConfig.getString(StreamsConfig.UPGRADE_FROM_CONFIG);
      if (upgradeFrom != null) {
        switch (upgradeFrom) {
          case StreamsConfig.UPGRADE_FROM_0100:
            log.info("Downgrading metadata version from {} to 1 for upgrade from 0.10.0.x.", SubscriptionInfo.LATEST_SUPPORTED_VERSION);
            usedSubscriptionMetadataVersion = VERSION_ONE;
            break;
          case StreamsConfig.UPGRADE_FROM_0101:
          case StreamsConfig.UPGRADE_FROM_0102:
          case StreamsConfig.UPGRADE_FROM_0110:
          case StreamsConfig.UPGRADE_FROM_10:
          case StreamsConfig.UPGRADE_FROM_11:
            log.info("Downgrading metadata version from {} to 2 for upgrade from {}.x.", SubscriptionInfo.LATEST_SUPPORTED_VERSION, upgradeFrom);
            usedSubscriptionMetadataVersion = VERSION_TWO;
            break;
          default:
            throw new IllegalArgumentException("Unknown configuration value for parameter 'upgrade.from': " + upgradeFrom);
        }
      }
  
      final Object o = configs.get(StreamsConfig.InternalConfig.TASK_MANAGER_FOR_PARTITION_ASSIGNOR);
      if (o == null) {
        final KafkaException fatalException = new KafkaException("TaskManager is not specified");
        log.error(fatalException.getMessage(), fatalException);
        throw fatalException;
      }
  
      if (!(o instanceof TaskManager)) {
        final KafkaException fatalException = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), TaskManager.class.getName()));
        log.error(fatalException.getMessage(), fatalException);
        throw fatalException;
      }
  
      taskManager = (TaskManager) o;
  
      final Object ai = configs.get(StreamsConfig.InternalConfig.ASSIGNMENT_ERROR_CODE);
      if (ai == null) {
        final KafkaException fatalException = new KafkaException("assignmentErrorCode is not specified");
        log.error(fatalException.getMessage(), fatalException);
        throw fatalException;
      }
  
      if (!(ai instanceof AtomicInteger)) {
        final KafkaException fatalException = new KafkaException(String.format("%s is not an instance of %s",
            ai.getClass().getName(), AtomicInteger.class.getName()));
        log.error(fatalException.getMessage(), fatalException);
        throw fatalException;
      }
      assignmentErrorCode = (AtomicInteger) ai;
  
      numStandbyReplicas = streamsConfig.getInt(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG);
  
      partitionGrouper = streamsConfig.getConfiguredInstance(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, PartitionGrouper.class);
  
      final String userEndPoint = streamsConfig.getString(StreamsConfig.APPLICATION_SERVER_CONFIG);
      if (userEndPoint != null && !userEndPoint.isEmpty()) {
        try {
          final String host = getHost(userEndPoint);
          final Integer port = getPort(userEndPoint);
  
          if (host == null || port == null)
            throw new ConfigException(String.format("%s Config %s isn't in the correct format. Expected a host:port pair" +
                    " but received %s",
                logPrefix, StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint));
        } catch (final NumberFormatException nfe) {
          throw new ConfigException(String.format("%s Invalid port supplied in %s for config %s",
              logPrefix, userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG));
        }
  
        this.userEndPoint = userEndPoint;
      }
  
      internalTopicManager = new InternalTopicManager(taskManager.adminClient, streamsConfig);
  
      copartitionedTopicsValidator = new CopartitionedTopicsValidator(logPrefix);
    }