Page History

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

            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<>();
            commandLineMap.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
            commandLineMap.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
            if (options.has(sendTimeoutOpt)) commandLineMap.put(LINGER_MS_CONFIG, options.valueOf(sendTimeoutOpt).toString());
            if (options.has(requestRequiredAcksOpt)) commandLineMap.put(ACKS_CONFIG, options.valueOf(requestRequiredAcksOpt).toString());
            if (options.has(requestTimeoutMsOpt)) commandLineMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valueOf(requestTimeoutMsOpt).toString());
            if (options.has(messageSendMaxRetriesOpt)) commandLineMap.put(RETRIES_CONFIG, options.valueOf(messageSendMaxRetriesOpt).toString());
            if (options.has(retryBackoffMsOpt)) commandLineMap.put(RETRY_BACKOFF_MS_CONFIG, options.valueOf(retryBackoffMsOpt).toString());
            if (options.has(socketBufferSizeOpt)) commandLineMap.put(SEND_BUFFER_CONFIG, options.valueOf(socketBufferSizeOpt).toString());
            if (options.has(maxMemoryBytesOpt)) commandLineMap.put(BUFFER_MEMORY_CONFIG, options.valueOf(maxMemoryBytesOpt).toString());
            if (options.has(batchSizeOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(batchSizeOpt).toString());
            if (options.has(maxPartitionMemoryBytesOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(maxPartitionMemoryBytesOpt).toString());
            if (options.has(metadataExpiryMsOpt)) commandLineMap.put(METADATA_MAX_AGE_CONFIG, options.valueOf(metadataExpiryMsOpt).toString());
            if (options.has(maxBlockMsOpt)) commandLineMap.put(MAX_BLOCK_MS_CONFIG, options.valueOf(maxBlockMsOpt).toString());

            // Properties passed as key=value pairs via command line
            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
            toolDefaultMap.put(LINGER_MS_CONFIG, options.valuesOf(sendTimeoutOpt).toString());
            toolDefaultMap.put(ACKS_CONFIG, options.valuesOf(requestRequiredAcksOpt).toString());
            toolDefaultMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valuesOf(requestTimeoutMsOpt).toString());
            toolDefaultMap.put(RETRIES_CONFIG, options.valuesOf(messageSendMaxRetriesOpt).toString());
            toolDefaultMap.put(RETRY_BACKOFF_MS_CONFIG, options.valuesOf(retryBackoffMsOpt).toString());
            toolDefaultMap.put(SEND_BUFFER_CONFIG, options.valuesOf(socketBufferSizeOpt).toString());
            toolDefaultMap.put(BUFFER_MEMORY_CONFIG, options.valuesOf(maxMemoryBytesOpt).toString());
            toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(batchSizeOpt).toString());
            toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(maxPartitionMemoryBytesOpt).toString());
            toolDefaultMap.put(METADATA_MAX_AGE_CONFIG, options.valuesOf(metadataExpiryMsOpt).toString());
            toolDefaultMap.put(MAX_BLOCK_MS_CONFIG, options.valuesOf(maxBlockMsOpt).toString());

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