/**
     * Merge multiple configuration sources according to priority, from highest to lowest:
     * 1) Command line arguments
     * 2) Properties passed as key=value pairs via command line (Command line properties)
     * 3) Configuration files
     * 4) Default values set by tool scripts and default values of command line arguments
     *
     * @param commandLineMap       Map of configuration from command line arguments
     * @param commandLineKeyValMap Map of configuration from key=value pairs passed via command line (command line properties)
     * @param configMap            Map of configuration from configuration files
     * @param toolDefaultMap       Map of default values provided by tool scripts or default command line options
     * @return a merged Map of all configuration sources, where higher-priority sources override lower-priority ones
     */
	public static Map<String, Object> mergePropertiesWithPrecedence(
        Map<String, Object> commandLineMap,
        Map<String, Object> commandLineKeyValMap,
        Map<String, Object> configMap,
        Map<String, Object> toolDefaultMap
    ) {
		warnConfigFromMultipleSources(commandLineMap, commandLineKeyValMap, configMap);
        
		Map<String, Object> map = new HashMap<>();
        // Default values set by tool scripts and default values of command line arguments
        if (toolDefaultMap != null) {
            map.putAll(toolDefaultMap);
        }
        // Configuration file
        if (configMap != null) {
            map.putAll(configMap);
        }
        // Properties passed as key=value pairs via command line (Command line properties)
        if (commandLineKeyValMap != null) {
            map.putAll(commandLineKeyValMap);
        }
        // Command line arguments
        if (commandLineMap != null) {
            map.putAll(commandLineMap);
        }
        return map;
    }

    /**
     * Print warnings when the same configurations is defined in multiple sources.
     * - Command line arguments override command property values and configuration files.
     * - Command property values override configuration files if no command line argument exists.
     *
     * @param commandLineMap       Map of configuration from command line arguments
     * @param commandLineKeyValMap Map of configuration from command line properties
     * @param configMap            Map of configuration loaded from configuration files
     */
    public static void warnConfigFromMultipleSources(
        Map<String, Object> commandLineMap,
        Map<String, Object> commandLineKeyValMap,
        Map<String, Object> configMap
    ) {
        // Command line argument overrides command property and config file
        for (Map.Entry<String, Object> entry : commandLineMap.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            StringBuilder warning = new StringBuilder();

            if (commandLineKeyValMap.containsKey(key) && configMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line argument overrides command property value (")
                    .append(commandLineKeyValMap.get(key)).append(")")
                    .append(" and configuration file value (")
                    .append(configMap.get(key)).append(").");
            } else if (commandLineKeyValMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line argument overrides command property value (")
                    .append(commandLineKeyValMap.get(key)).append(").");
            } else if (configMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line overrides configuration file value (")
                    .append(configMap.get(key)).append(").");
            }

            if (!warning.isEmpty()) {
                System.out.println("Warning: " + warning);
            }
        }

        // Command property override config file
        for (Map.Entry<String, Object> entry : commandLineKeyValMap.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            if (commandLineMap.containsKey(key)) continue;

            if (configMap.containsKey(key)) {
                System.out.println("Warning: " + key + "=" + value
                    + " from command property overrides configuration file value ("
                    + configMap.get(key) + ").");
            }
        }
    }

	// joptsimple-specific helper method
    public static <T> void putIfOption(OptionSet options, OptionSpec<T> spec, Map<String, Object> map, String key) {
        if (options.has(spec)) {
            T value = options.valueOf(spec);
            map.put(key, value);
        }
    }

	// joptsimple-specific helper method
    public static <T> void putOption(OptionSet options, OptionSpec<T> spec, Map<String, Object> map, String key) {
        T value = options.valueOf(spec);
        map.put(key, value);
    }

        Map<String, Object> readerProps() throws IOException {
            Map<String, Object> commandLineMap = new HashMap<>();
            putOption(options, topicOpt, commandLineMap, "topic");

            Map<String, Object> configMap = new HashMap<>();
            if (options.has(readerConfigOpt)) {
                configMap.putAll(propsToStringMap(loadProps(options.valueOf(readerConfigOpt))));
            }

            Map<String, Object> commandLineKeyValMap = new HashMap<>();
            if (options.has(readerPropertyOpt)) {
                commandLineKeyValMap.putAll(propsToStringMap(parseKeyValueArgs(options.valuesOf(readerPropertyOpt))));
            }
            return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, null);
        }
        
		Map<String, Object> producerProps() throws IOException {
            // Prepare the map from command line arguments
            Map<String, Object>  commandLineMap = new HashMap<>();
            putIfOption(options, bootstrapServerOpt, commandLineMap, BOOTSTRAP_SERVERS_CONFIG);
            commandLineMap.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
            putIfOption(options, sendTimeoutOpt, commandLineMap, LINGER_MS_CONFIG);
            putIfOption(options, requestRequiredAcksOpt, commandLineMap, ACKS_CONFIG);
            putIfOption(options, requestTimeoutMsOpt, commandLineMap, REQUEST_TIMEOUT_MS_CONFIG);
            putIfOption(options, messageSendMaxRetriesOpt, commandLineMap, RETRIES_CONFIG);
            putIfOption(options, retryBackoffMsOpt, commandLineMap, RETRY_BACKOFF_MS_CONFIG);
            putIfOption(options, socketBufferSizeOpt, commandLineMap, SEND_BUFFER_CONFIG);
            putIfOption(options, maxMemoryBytesOpt, commandLineMap, BUFFER_MEMORY_CONFIG);
            // We currently have 2 options to set the batch.size value. We'll deprecate/remove one of them in KIP-717.
            putIfOption(options, batchSizeOpt, commandLineMap, BATCH_SIZE_CONFIG);
            putIfOption(options, maxPartitionMemoryBytesOpt, commandLineMap, BATCH_SIZE_CONFIG);
            putIfOption(options, metadataExpiryMsOpt, commandLineMap, METADATA_MAX_AGE_CONFIG);
            putIfOption(options, maxBlockMsOpt, commandLineMap, MAX_BLOCK_MS_CONFIG);

            // Properties passed as key=value pairs via command line (Command line properties)
            Map<String, Object> commandLineKeyValMap = new HashMap<>();
            if (options.has(commandPropertyOpt)) {
                commandLineKeyValMap.putAll(Utils.propsToStringMap(
                        parseKeyValueArgs(options.valuesOf(commandPropertyOpt))
                ));
            }
            // Configuration files
            Map<String, Object> configMap = new HashMap<>();
            if (options.has(commandConfigOpt)) {
                configMap.putAll(Utils.propsToStringMap(
                        Utils.loadProps(options.valueOf(commandConfigOpt))
                ));
            }

            // Default values set by tool scripts and default values of command line arguments
            Map<String, Object> toolDefaultMap = new HashMap<>();
            toolDefaultMap.put(CLIENT_ID_CONFIG, "console-producer");
            toolDefaultMap.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
            toolDefaultMap.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
            // Since all the options below have default values, we don't need to check for null
            putOption(options, sendTimeoutOpt, toolDefaultMap, LINGER_MS_CONFIG);
            putOption(options, requestRequiredAcksOpt, toolDefaultMap, ACKS_CONFIG);
            putOption(options, requestTimeoutMsOpt, toolDefaultMap, REQUEST_TIMEOUT_MS_CONFIG);
            putOption(options, messageSendMaxRetriesOpt, toolDefaultMap, RETRIES_CONFIG);
            putOption(options, retryBackoffMsOpt, toolDefaultMap, RETRY_BACKOFF_MS_CONFIG);
            putOption(options, socketBufferSizeOpt, toolDefaultMap, SEND_BUFFER_CONFIG);
            putOption(options, maxMemoryBytesOpt, toolDefaultMap, BUFFER_MEMORY_CONFIG);
            putOption(options, batchSizeOpt, toolDefaultMap, BATCH_SIZE_CONFIG);
            putOption(options, maxPartitionMemoryBytesOpt, toolDefaultMap, BATCH_SIZE_CONFIG);
            putOption(options, metadataExpiryMsOpt, toolDefaultMap, METADATA_MAX_AGE_CONFIG);
            putOption(options, maxBlockMsOpt, toolDefaultMap, MAX_BLOCK_MS_CONFIG);

			return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, toolDefaultMap);
        }