private void validateRackAwarenessConfiguration() {
        final List<String> rackAwareAssignmentTags = getList(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
        final Map<String, String> clientTags = getClientTags();

        if (clientTags.size() > MAX_RACK_AWARE_ASSIGNMENT_TAG_LIST_SIZE) {
            throw new ConfigException("At most " + MAX_RACK_AWARE_ASSIGNMENT_TAG_LIST_SIZE + " client tags " +
                                      "can be specified using " + CLIENT_TAG_PREFIX + " prefix.");
        }

        for (final String rackAwareAssignmentTag : rackAwareAssignmentTags) {
            if (!clientTags.containsKey(rackAwareAssignmentTag)) {
                throw new ConfigException(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
                                          rackAwareAssignmentTags,
                                          "Contains invalid value [" + rackAwareAssignmentTag + "] " +
                                          "which doesn't have corresponding tag set via [" + CLIENT_TAG_PREFIX + "] prefix.");
            }
        }

        clientTags.forEach((tagKey, tagValue) -> {
            if (tagKey.length() > MAX_RACK_AWARE_ASSIGNMENT_TAG_KEY_LENGTH) {
                throw new ConfigException(CLIENT_TAG_PREFIX,
                                          tagKey,
                                          "Tag key exceeds maximum length of " + MAX_RACK_AWARE_ASSIGNMENT_TAG_KEY_LENGTH + ".");
            }
            if (tagValue.length() > MAX_RACK_AWARE_ASSIGNMENT_TAG_VALUE_LENGTH) {
                throw new ConfigException(CLIENT_TAG_PREFIX,
                                          tagValue,
                                          "Tag value exceeds maximum length of " + MAX_RACK_AWARE_ASSIGNMENT_TAG_VALUE_LENGTH + ".");
            }
        });
    }
Create a new {@code StreamsConfig} using the given properties.

@param props properties that specify Kafka Streams and internal consumer/producer configuration
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java