@Test public void testKafkaProducerConfigAck() throws ConfigurationException { Map<String, Object> map = new HashMap<>(); map.put("bootstrap.servers", "localhost:9092"); AbstractConfiguration config = new MapConfiguration(map); try { KafkaProducerConfig producerConfig = LogConfigUtils.parseProducerConfig(config); assertEquals(producerConfig.getAcks(), "1"); } catch (ConfigurationException e) { throw e; } map.put(SingerConfigDef.REQUEST_REQUIRED_ACKS, "1"); try { KafkaProducerConfig producerConfig = LogConfigUtils.parseProducerConfig(config); assertEquals(producerConfig.getAcks(), "1"); } catch (ConfigurationException e) { throw e; } map.put(SingerConfigDef.REQUEST_REQUIRED_ACKS, "-1"); config = new MapConfiguration(map); try { KafkaProducerConfig producerConfig = LogConfigUtils.parseProducerConfig(config); assertEquals(producerConfig.getAcks(), "-1"); KafkaUtils.createKafkaProducer(producerConfig); } catch (ConfigurationException e) { throw e; } map.put(SingerConfigDef.REQUEST_REQUIRED_ACKS, "2"); try { LogConfigUtils.parseProducerConfig(config); fail("Must have thrown an exception since invalid ack config was passed"); } catch (ConfigurationException e) { } map.put(SingerConfigDef.ACKS, "all"); config = new MapConfiguration(map); try { KafkaProducerConfig producerConfig = LogConfigUtils.parseProducerConfig(config); assertEquals(producerConfig.getAcks(), "all"); KafkaUtils.createKafkaProducer(producerConfig); } catch (ConfigurationException e) { throw e; } }
public static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration) throws ConfigurationException { producerConfiguration.setThrowExceptionOnMissing(false); String partitionClass = producerConfiguration.getString(SingerConfigDef.PARTITIONER_CLASS); Set<String> brokerSet = Sets .newHashSet(producerConfiguration.getStringArray(SingerConfigDef.BOOTSTRAP_SERVERS)); String serverSetFilePath = null; if (producerConfiguration.containsKey(SingerConfigDef.BOOTSTRAP_SERVERS_FILE)) { serverSetFilePath = producerConfiguration.getString(SingerConfigDef.BOOTSTRAP_SERVERS_FILE); } else if (producerConfiguration.containsKey(SingerConfigDef.BROKER_SERVERSET_DEPRECATED)) { String serversetZkPath = producerConfiguration .getString(SingerConfigDef.BROKER_SERVERSET_DEPRECATED); if (SHADOW_MODE_ENABLED) { serversetZkPath = SHADOW_SERVERSET_MAPPING.getProperty(serversetZkPath, SHADOW_SERVERSET_MAPPING.getProperty(DEFAULT_SHADOW_SERVERSET)); } serverSetFilePath = filePathFromZKPath(serversetZkPath); } if (brokerSet.isEmpty() && Strings.isNullOrEmpty(serverSetFilePath)) { throw new ConfigurationException( "metadata.broker.list or metadata.broker.serverset needs to be set."); } if (brokerSet.isEmpty()) { if (!KAFKA_SERVER_SETS.containsKey(serverSetFilePath)) { try { ConfigFileServerSet serverSet = new ConfigFileServerSet(serverSetFilePath); final String monitoredServersetFilePath = serverSetFilePath; serverSet .monitor(new BrokerSetChangeListener(monitoredServersetFilePath, KAFKA_SERVER_SETS)); } catch (Exception e) { throw new ConfigurationException("Cannot get broker list from serverset.", e); } } LOG.debug("Initial loading kafka broker serverset OK."); brokerSet = KAFKA_SERVER_SETS.get(serverSetFilePath); } else if (!Strings.isNullOrEmpty(serverSetFilePath)) { LOG.warn("Ignoring metadata.broker.serverset when metadata.broker.list is configured."); } producerConfiguration.setThrowExceptionOnMissing(true); String acks = producerConfiguration.containsKey(SingerConfigDef.ACKS) ? producerConfiguration.getString(SingerConfigDef.ACKS) : producerConfiguration.getString(SingerConfigDef.REQUEST_REQUIRED_ACKS, DEFAULT_ACKS); acks = acks.toLowerCase(); if (!acks.equals(ACKS_ALL) && Math.abs(Integer.parseInt(acks)) > Integer.valueOf(DEFAULT_ACKS)) { throw new ConfigurationException( "Invalid ack configuration, " + SingerConfigDef.ACKS + "configuration value can only be 'all', -1, 0, 1"); } KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(serverSetFilePath, Lists.newArrayList(brokerSet), acks); if (partitionClass != null) { try { Class.forName(partitionClass); } catch (ClassNotFoundException e) { throw new ConfigurationException( "Unknown partitioner class: " + partitionClass); } kafkaProducerConfig.setPartitionerClass(partitionClass); } else { kafkaProducerConfig.setPartitionerClass(SingerConfigDef.DEFAULT_PARTITIONER); } if (producerConfiguration.containsKey(SingerConfigDef.COMPRESSION_TYPE)) { String compressionType = producerConfiguration.getString(SingerConfigDef.COMPRESSION_TYPE); if (compressionType != null && !compressionType.isEmpty()) { try { CompressionType.forName(compressionType); }catch(Exception e) { throw new ConfigurationException("Unknown compression type: " + compressionType); } kafkaProducerConfig.setCompressionType(compressionType); } } if (producerConfiguration.containsKey(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)) { int maxRequestSize = producerConfiguration.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); kafkaProducerConfig.setMaxRequestSize(maxRequestSize); } if (producerConfiguration.containsKey(SingerConfigDef.SSL_ENABLED_CONFIG)) { boolean enabled = producerConfiguration.getBoolean(SingerConfigDef.SSL_ENABLED_CONFIG); kafkaProducerConfig.setSslEnabled(enabled); if (enabled) { List<String> brokers = kafkaProducerConfig.getBrokerLists(); List<String> updated = new ArrayList<>(); for (int i = 0; i < brokers.size(); i++) { String broker = brokers.get(i); String[] ipports = broker.split(":"); try { InetAddress addr = InetAddress.getByName(ipports[0]); String host = addr.getHostName(); String newBrokerStr = host + ":9093"; updated.add(newBrokerStr); } catch (UnknownHostException e) { LOG.error("Unknown host: {}", ipports[0]); } } kafkaProducerConfig.setBrokerLists(updated); Iterator<String> sslKeysIterator = producerConfiguration.getKeys("ssl"); kafkaProducerConfig.setSslSettings(new HashMap<>()); while (sslKeysIterator.hasNext()) { String key = sslKeysIterator.next(); String value = key.equals(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) ? String.join(",", producerConfiguration.getStringArray(key)) : producerConfiguration.getString(key); kafkaProducerConfig.getSslSettings().put(key, value); } } } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_ENABLED_CONFIG)) { kafkaProducerConfig.setTransactionEnabled(true); } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG)) { int timeoutMs = producerConfiguration.getInt(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG); kafkaProducerConfig.setTransactionTimeoutMs(timeoutMs); } if (producerConfiguration.containsKey(SingerConfigDef.RETRIES_CONFIG)) { int retries = producerConfiguration.getInt(SingerConfigDef.RETRIES_CONFIG); kafkaProducerConfig.setRetries(retries); } if (producerConfiguration.containsKey(SingerConfigDef.PRODUCER_BUFFER_MEMORY)) { int bufferMemory = producerConfiguration.getInt(SingerConfigDef.PRODUCER_BUFFER_MEMORY); kafkaProducerConfig.setBufferMemory(bufferMemory); } return kafkaProducerConfig; }
LogConfigUtils { public static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration) throws ConfigurationException { producerConfiguration.setThrowExceptionOnMissing(false); String partitionClass = producerConfiguration.getString(SingerConfigDef.PARTITIONER_CLASS); Set<String> brokerSet = Sets .newHashSet(producerConfiguration.getStringArray(SingerConfigDef.BOOTSTRAP_SERVERS)); String serverSetFilePath = null; if (producerConfiguration.containsKey(SingerConfigDef.BOOTSTRAP_SERVERS_FILE)) { serverSetFilePath = producerConfiguration.getString(SingerConfigDef.BOOTSTRAP_SERVERS_FILE); } else if (producerConfiguration.containsKey(SingerConfigDef.BROKER_SERVERSET_DEPRECATED)) { String serversetZkPath = producerConfiguration .getString(SingerConfigDef.BROKER_SERVERSET_DEPRECATED); if (SHADOW_MODE_ENABLED) { serversetZkPath = SHADOW_SERVERSET_MAPPING.getProperty(serversetZkPath, SHADOW_SERVERSET_MAPPING.getProperty(DEFAULT_SHADOW_SERVERSET)); } serverSetFilePath = filePathFromZKPath(serversetZkPath); } if (brokerSet.isEmpty() && Strings.isNullOrEmpty(serverSetFilePath)) { throw new ConfigurationException( "metadata.broker.list or metadata.broker.serverset needs to be set."); } if (brokerSet.isEmpty()) { if (!KAFKA_SERVER_SETS.containsKey(serverSetFilePath)) { try { ConfigFileServerSet serverSet = new ConfigFileServerSet(serverSetFilePath); final String monitoredServersetFilePath = serverSetFilePath; serverSet .monitor(new BrokerSetChangeListener(monitoredServersetFilePath, KAFKA_SERVER_SETS)); } catch (Exception e) { throw new ConfigurationException("Cannot get broker list from serverset.", e); } } LOG.debug("Initial loading kafka broker serverset OK."); brokerSet = KAFKA_SERVER_SETS.get(serverSetFilePath); } else if (!Strings.isNullOrEmpty(serverSetFilePath)) { LOG.warn("Ignoring metadata.broker.serverset when metadata.broker.list is configured."); } producerConfiguration.setThrowExceptionOnMissing(true); String acks = producerConfiguration.containsKey(SingerConfigDef.ACKS) ? producerConfiguration.getString(SingerConfigDef.ACKS) : producerConfiguration.getString(SingerConfigDef.REQUEST_REQUIRED_ACKS, DEFAULT_ACKS); acks = acks.toLowerCase(); if (!acks.equals(ACKS_ALL) && Math.abs(Integer.parseInt(acks)) > Integer.valueOf(DEFAULT_ACKS)) { throw new ConfigurationException( "Invalid ack configuration, " + SingerConfigDef.ACKS + "configuration value can only be 'all', -1, 0, 1"); } KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(serverSetFilePath, Lists.newArrayList(brokerSet), acks); if (partitionClass != null) { try { Class.forName(partitionClass); } catch (ClassNotFoundException e) { throw new ConfigurationException( "Unknown partitioner class: " + partitionClass); } kafkaProducerConfig.setPartitionerClass(partitionClass); } else { kafkaProducerConfig.setPartitionerClass(SingerConfigDef.DEFAULT_PARTITIONER); } if (producerConfiguration.containsKey(SingerConfigDef.COMPRESSION_TYPE)) { String compressionType = producerConfiguration.getString(SingerConfigDef.COMPRESSION_TYPE); if (compressionType != null && !compressionType.isEmpty()) { try { CompressionType.forName(compressionType); }catch(Exception e) { throw new ConfigurationException("Unknown compression type: " + compressionType); } kafkaProducerConfig.setCompressionType(compressionType); } } if (producerConfiguration.containsKey(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)) { int maxRequestSize = producerConfiguration.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); kafkaProducerConfig.setMaxRequestSize(maxRequestSize); } if (producerConfiguration.containsKey(SingerConfigDef.SSL_ENABLED_CONFIG)) { boolean enabled = producerConfiguration.getBoolean(SingerConfigDef.SSL_ENABLED_CONFIG); kafkaProducerConfig.setSslEnabled(enabled); if (enabled) { List<String> brokers = kafkaProducerConfig.getBrokerLists(); List<String> updated = new ArrayList<>(); for (int i = 0; i < brokers.size(); i++) { String broker = brokers.get(i); String[] ipports = broker.split(":"); try { InetAddress addr = InetAddress.getByName(ipports[0]); String host = addr.getHostName(); String newBrokerStr = host + ":9093"; updated.add(newBrokerStr); } catch (UnknownHostException e) { LOG.error("Unknown host: {}", ipports[0]); } } kafkaProducerConfig.setBrokerLists(updated); Iterator<String> sslKeysIterator = producerConfiguration.getKeys("ssl"); kafkaProducerConfig.setSslSettings(new HashMap<>()); while (sslKeysIterator.hasNext()) { String key = sslKeysIterator.next(); String value = key.equals(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) ? String.join(",", producerConfiguration.getStringArray(key)) : producerConfiguration.getString(key); kafkaProducerConfig.getSslSettings().put(key, value); } } } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_ENABLED_CONFIG)) { kafkaProducerConfig.setTransactionEnabled(true); } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG)) { int timeoutMs = producerConfiguration.getInt(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG); kafkaProducerConfig.setTransactionTimeoutMs(timeoutMs); } if (producerConfiguration.containsKey(SingerConfigDef.RETRIES_CONFIG)) { int retries = producerConfiguration.getInt(SingerConfigDef.RETRIES_CONFIG); kafkaProducerConfig.setRetries(retries); } if (producerConfiguration.containsKey(SingerConfigDef.PRODUCER_BUFFER_MEMORY)) { int bufferMemory = producerConfiguration.getInt(SingerConfigDef.PRODUCER_BUFFER_MEMORY); kafkaProducerConfig.setBufferMemory(bufferMemory); } return kafkaProducerConfig; } }
LogConfigUtils { public static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration) throws ConfigurationException { producerConfiguration.setThrowExceptionOnMissing(false); String partitionClass = producerConfiguration.getString(SingerConfigDef.PARTITIONER_CLASS); Set<String> brokerSet = Sets .newHashSet(producerConfiguration.getStringArray(SingerConfigDef.BOOTSTRAP_SERVERS)); String serverSetFilePath = null; if (producerConfiguration.containsKey(SingerConfigDef.BOOTSTRAP_SERVERS_FILE)) { serverSetFilePath = producerConfiguration.getString(SingerConfigDef.BOOTSTRAP_SERVERS_FILE); } else if (producerConfiguration.containsKey(SingerConfigDef.BROKER_SERVERSET_DEPRECATED)) { String serversetZkPath = producerConfiguration .getString(SingerConfigDef.BROKER_SERVERSET_DEPRECATED); if (SHADOW_MODE_ENABLED) { serversetZkPath = SHADOW_SERVERSET_MAPPING.getProperty(serversetZkPath, SHADOW_SERVERSET_MAPPING.getProperty(DEFAULT_SHADOW_SERVERSET)); } serverSetFilePath = filePathFromZKPath(serversetZkPath); } if (brokerSet.isEmpty() && Strings.isNullOrEmpty(serverSetFilePath)) { throw new ConfigurationException( "metadata.broker.list or metadata.broker.serverset needs to be set."); } if (brokerSet.isEmpty()) { if (!KAFKA_SERVER_SETS.containsKey(serverSetFilePath)) { try { ConfigFileServerSet serverSet = new ConfigFileServerSet(serverSetFilePath); final String monitoredServersetFilePath = serverSetFilePath; serverSet .monitor(new BrokerSetChangeListener(monitoredServersetFilePath, KAFKA_SERVER_SETS)); } catch (Exception e) { throw new ConfigurationException("Cannot get broker list from serverset.", e); } } LOG.debug("Initial loading kafka broker serverset OK."); brokerSet = KAFKA_SERVER_SETS.get(serverSetFilePath); } else if (!Strings.isNullOrEmpty(serverSetFilePath)) { LOG.warn("Ignoring metadata.broker.serverset when metadata.broker.list is configured."); } producerConfiguration.setThrowExceptionOnMissing(true); String acks = producerConfiguration.containsKey(SingerConfigDef.ACKS) ? producerConfiguration.getString(SingerConfigDef.ACKS) : producerConfiguration.getString(SingerConfigDef.REQUEST_REQUIRED_ACKS, DEFAULT_ACKS); acks = acks.toLowerCase(); if (!acks.equals(ACKS_ALL) && Math.abs(Integer.parseInt(acks)) > Integer.valueOf(DEFAULT_ACKS)) { throw new ConfigurationException( "Invalid ack configuration, " + SingerConfigDef.ACKS + "configuration value can only be 'all', -1, 0, 1"); } KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(serverSetFilePath, Lists.newArrayList(brokerSet), acks); if (partitionClass != null) { try { Class.forName(partitionClass); } catch (ClassNotFoundException e) { throw new ConfigurationException( "Unknown partitioner class: " + partitionClass); } kafkaProducerConfig.setPartitionerClass(partitionClass); } else { kafkaProducerConfig.setPartitionerClass(SingerConfigDef.DEFAULT_PARTITIONER); } if (producerConfiguration.containsKey(SingerConfigDef.COMPRESSION_TYPE)) { String compressionType = producerConfiguration.getString(SingerConfigDef.COMPRESSION_TYPE); if (compressionType != null && !compressionType.isEmpty()) { try { CompressionType.forName(compressionType); }catch(Exception e) { throw new ConfigurationException("Unknown compression type: " + compressionType); } kafkaProducerConfig.setCompressionType(compressionType); } } if (producerConfiguration.containsKey(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)) { int maxRequestSize = producerConfiguration.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); kafkaProducerConfig.setMaxRequestSize(maxRequestSize); } if (producerConfiguration.containsKey(SingerConfigDef.SSL_ENABLED_CONFIG)) { boolean enabled = producerConfiguration.getBoolean(SingerConfigDef.SSL_ENABLED_CONFIG); kafkaProducerConfig.setSslEnabled(enabled); if (enabled) { List<String> brokers = kafkaProducerConfig.getBrokerLists(); List<String> updated = new ArrayList<>(); for (int i = 0; i < brokers.size(); i++) { String broker = brokers.get(i); String[] ipports = broker.split(":"); try { InetAddress addr = InetAddress.getByName(ipports[0]); String host = addr.getHostName(); String newBrokerStr = host + ":9093"; updated.add(newBrokerStr); } catch (UnknownHostException e) { LOG.error("Unknown host: {}", ipports[0]); } } kafkaProducerConfig.setBrokerLists(updated); Iterator<String> sslKeysIterator = producerConfiguration.getKeys("ssl"); kafkaProducerConfig.setSslSettings(new HashMap<>()); while (sslKeysIterator.hasNext()) { String key = sslKeysIterator.next(); String value = key.equals(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) ? String.join(",", producerConfiguration.getStringArray(key)) : producerConfiguration.getString(key); kafkaProducerConfig.getSslSettings().put(key, value); } } } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_ENABLED_CONFIG)) { kafkaProducerConfig.setTransactionEnabled(true); } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG)) { int timeoutMs = producerConfiguration.getInt(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG); kafkaProducerConfig.setTransactionTimeoutMs(timeoutMs); } if (producerConfiguration.containsKey(SingerConfigDef.RETRIES_CONFIG)) { int retries = producerConfiguration.getInt(SingerConfigDef.RETRIES_CONFIG); kafkaProducerConfig.setRetries(retries); } if (producerConfiguration.containsKey(SingerConfigDef.PRODUCER_BUFFER_MEMORY)) { int bufferMemory = producerConfiguration.getInt(SingerConfigDef.PRODUCER_BUFFER_MEMORY); kafkaProducerConfig.setBufferMemory(bufferMemory); } return kafkaProducerConfig; } private  LogConfigUtils(); }
LogConfigUtils { public static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration) throws ConfigurationException { producerConfiguration.setThrowExceptionOnMissing(false); String partitionClass = producerConfiguration.getString(SingerConfigDef.PARTITIONER_CLASS); Set<String> brokerSet = Sets .newHashSet(producerConfiguration.getStringArray(SingerConfigDef.BOOTSTRAP_SERVERS)); String serverSetFilePath = null; if (producerConfiguration.containsKey(SingerConfigDef.BOOTSTRAP_SERVERS_FILE)) { serverSetFilePath = producerConfiguration.getString(SingerConfigDef.BOOTSTRAP_SERVERS_FILE); } else if (producerConfiguration.containsKey(SingerConfigDef.BROKER_SERVERSET_DEPRECATED)) { String serversetZkPath = producerConfiguration .getString(SingerConfigDef.BROKER_SERVERSET_DEPRECATED); if (SHADOW_MODE_ENABLED) { serversetZkPath = SHADOW_SERVERSET_MAPPING.getProperty(serversetZkPath, SHADOW_SERVERSET_MAPPING.getProperty(DEFAULT_SHADOW_SERVERSET)); } serverSetFilePath = filePathFromZKPath(serversetZkPath); } if (brokerSet.isEmpty() && Strings.isNullOrEmpty(serverSetFilePath)) { throw new ConfigurationException( "metadata.broker.list or metadata.broker.serverset needs to be set."); } if (brokerSet.isEmpty()) { if (!KAFKA_SERVER_SETS.containsKey(serverSetFilePath)) { try { ConfigFileServerSet serverSet = new ConfigFileServerSet(serverSetFilePath); final String monitoredServersetFilePath = serverSetFilePath; serverSet .monitor(new BrokerSetChangeListener(monitoredServersetFilePath, KAFKA_SERVER_SETS)); } catch (Exception e) { throw new ConfigurationException("Cannot get broker list from serverset.", e); } } LOG.debug("Initial loading kafka broker serverset OK."); brokerSet = KAFKA_SERVER_SETS.get(serverSetFilePath); } else if (!Strings.isNullOrEmpty(serverSetFilePath)) { LOG.warn("Ignoring metadata.broker.serverset when metadata.broker.list is configured."); } producerConfiguration.setThrowExceptionOnMissing(true); String acks = producerConfiguration.containsKey(SingerConfigDef.ACKS) ? producerConfiguration.getString(SingerConfigDef.ACKS) : producerConfiguration.getString(SingerConfigDef.REQUEST_REQUIRED_ACKS, DEFAULT_ACKS); acks = acks.toLowerCase(); if (!acks.equals(ACKS_ALL) && Math.abs(Integer.parseInt(acks)) > Integer.valueOf(DEFAULT_ACKS)) { throw new ConfigurationException( "Invalid ack configuration, " + SingerConfigDef.ACKS + "configuration value can only be 'all', -1, 0, 1"); } KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(serverSetFilePath, Lists.newArrayList(brokerSet), acks); if (partitionClass != null) { try { Class.forName(partitionClass); } catch (ClassNotFoundException e) { throw new ConfigurationException( "Unknown partitioner class: " + partitionClass); } kafkaProducerConfig.setPartitionerClass(partitionClass); } else { kafkaProducerConfig.setPartitionerClass(SingerConfigDef.DEFAULT_PARTITIONER); } if (producerConfiguration.containsKey(SingerConfigDef.COMPRESSION_TYPE)) { String compressionType = producerConfiguration.getString(SingerConfigDef.COMPRESSION_TYPE); if (compressionType != null && !compressionType.isEmpty()) { try { CompressionType.forName(compressionType); }catch(Exception e) { throw new ConfigurationException("Unknown compression type: " + compressionType); } kafkaProducerConfig.setCompressionType(compressionType); } } if (producerConfiguration.containsKey(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)) { int maxRequestSize = producerConfiguration.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); kafkaProducerConfig.setMaxRequestSize(maxRequestSize); } if (producerConfiguration.containsKey(SingerConfigDef.SSL_ENABLED_CONFIG)) { boolean enabled = producerConfiguration.getBoolean(SingerConfigDef.SSL_ENABLED_CONFIG); kafkaProducerConfig.setSslEnabled(enabled); if (enabled) { List<String> brokers = kafkaProducerConfig.getBrokerLists(); List<String> updated = new ArrayList<>(); for (int i = 0; i < brokers.size(); i++) { String broker = brokers.get(i); String[] ipports = broker.split(":"); try { InetAddress addr = InetAddress.getByName(ipports[0]); String host = addr.getHostName(); String newBrokerStr = host + ":9093"; updated.add(newBrokerStr); } catch (UnknownHostException e) { LOG.error("Unknown host: {}", ipports[0]); } } kafkaProducerConfig.setBrokerLists(updated); Iterator<String> sslKeysIterator = producerConfiguration.getKeys("ssl"); kafkaProducerConfig.setSslSettings(new HashMap<>()); while (sslKeysIterator.hasNext()) { String key = sslKeysIterator.next(); String value = key.equals(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) ? String.join(",", producerConfiguration.getStringArray(key)) : producerConfiguration.getString(key); kafkaProducerConfig.getSslSettings().put(key, value); } } } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_ENABLED_CONFIG)) { kafkaProducerConfig.setTransactionEnabled(true); } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG)) { int timeoutMs = producerConfiguration.getInt(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG); kafkaProducerConfig.setTransactionTimeoutMs(timeoutMs); } if (producerConfiguration.containsKey(SingerConfigDef.RETRIES_CONFIG)) { int retries = producerConfiguration.getInt(SingerConfigDef.RETRIES_CONFIG); kafkaProducerConfig.setRetries(retries); } if (producerConfiguration.containsKey(SingerConfigDef.PRODUCER_BUFFER_MEMORY)) { int bufferMemory = producerConfiguration.getInt(SingerConfigDef.PRODUCER_BUFFER_MEMORY); kafkaProducerConfig.setBufferMemory(bufferMemory); } return kafkaProducerConfig; } private  LogConfigUtils(); static SingerConfig parseDirBasedSingerConfigHeader(String singerConfigFile); static MemqWriterConfig parseMemqWriterConfig(SubsetConfiguration configuration); static void parseMemqAuditorConfigs(MemqWriterConfig config,
                                         AbstractConfiguration configuration); static SingerConfig parseFileBasedSingerConfig(String singerConfigFile); static LogMonitorConfig parseLogMonitorConfig(AbstractConfiguration monitorConfiguration); static SingerLogConfig parseLogConfigFromFile(File logConfigFile); static SingerLogConfig[] parseLogStreamConfigFromFile(String logConfigFile); static String parseLogNamespace(File logConfigFile); static SingerLogConfig parseLogConfig(String logName,
                                               AbstractConfiguration logConfiguration); static SingerLogConfig[] parseLogStreamConfig(AbstractConfiguration config); static SingerConfig parseCommonSingerConfigHeader(AbstractConfiguration singerConfiguration); static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration); static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration); static String filePathFromZKPath(String serverSetZKPath); static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration); static PropertiesConfiguration[] getTopicConfigs(Configuration config); static PropertiesConfiguration getTopicConfig(Configuration config, String topicName); static void validateNewConfig(Configuration config); }
LogConfigUtils { public static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration) throws ConfigurationException { producerConfiguration.setThrowExceptionOnMissing(false); String partitionClass = producerConfiguration.getString(SingerConfigDef.PARTITIONER_CLASS); Set<String> brokerSet = Sets .newHashSet(producerConfiguration.getStringArray(SingerConfigDef.BOOTSTRAP_SERVERS)); String serverSetFilePath = null; if (producerConfiguration.containsKey(SingerConfigDef.BOOTSTRAP_SERVERS_FILE)) { serverSetFilePath = producerConfiguration.getString(SingerConfigDef.BOOTSTRAP_SERVERS_FILE); } else if (producerConfiguration.containsKey(SingerConfigDef.BROKER_SERVERSET_DEPRECATED)) { String serversetZkPath = producerConfiguration .getString(SingerConfigDef.BROKER_SERVERSET_DEPRECATED); if (SHADOW_MODE_ENABLED) { serversetZkPath = SHADOW_SERVERSET_MAPPING.getProperty(serversetZkPath, SHADOW_SERVERSET_MAPPING.getProperty(DEFAULT_SHADOW_SERVERSET)); } serverSetFilePath = filePathFromZKPath(serversetZkPath); } if (brokerSet.isEmpty() && Strings.isNullOrEmpty(serverSetFilePath)) { throw new ConfigurationException( "metadata.broker.list or metadata.broker.serverset needs to be set."); } if (brokerSet.isEmpty()) { if (!KAFKA_SERVER_SETS.containsKey(serverSetFilePath)) { try { ConfigFileServerSet serverSet = new ConfigFileServerSet(serverSetFilePath); final String monitoredServersetFilePath = serverSetFilePath; serverSet .monitor(new BrokerSetChangeListener(monitoredServersetFilePath, KAFKA_SERVER_SETS)); } catch (Exception e) { throw new ConfigurationException("Cannot get broker list from serverset.", e); } } LOG.debug("Initial loading kafka broker serverset OK."); brokerSet = KAFKA_SERVER_SETS.get(serverSetFilePath); } else if (!Strings.isNullOrEmpty(serverSetFilePath)) { LOG.warn("Ignoring metadata.broker.serverset when metadata.broker.list is configured."); } producerConfiguration.setThrowExceptionOnMissing(true); String acks = producerConfiguration.containsKey(SingerConfigDef.ACKS) ? producerConfiguration.getString(SingerConfigDef.ACKS) : producerConfiguration.getString(SingerConfigDef.REQUEST_REQUIRED_ACKS, DEFAULT_ACKS); acks = acks.toLowerCase(); if (!acks.equals(ACKS_ALL) && Math.abs(Integer.parseInt(acks)) > Integer.valueOf(DEFAULT_ACKS)) { throw new ConfigurationException( "Invalid ack configuration, " + SingerConfigDef.ACKS + "configuration value can only be 'all', -1, 0, 1"); } KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(serverSetFilePath, Lists.newArrayList(brokerSet), acks); if (partitionClass != null) { try { Class.forName(partitionClass); } catch (ClassNotFoundException e) { throw new ConfigurationException( "Unknown partitioner class: " + partitionClass); } kafkaProducerConfig.setPartitionerClass(partitionClass); } else { kafkaProducerConfig.setPartitionerClass(SingerConfigDef.DEFAULT_PARTITIONER); } if (producerConfiguration.containsKey(SingerConfigDef.COMPRESSION_TYPE)) { String compressionType = producerConfiguration.getString(SingerConfigDef.COMPRESSION_TYPE); if (compressionType != null && !compressionType.isEmpty()) { try { CompressionType.forName(compressionType); }catch(Exception e) { throw new ConfigurationException("Unknown compression type: " + compressionType); } kafkaProducerConfig.setCompressionType(compressionType); } } if (producerConfiguration.containsKey(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)) { int maxRequestSize = producerConfiguration.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); kafkaProducerConfig.setMaxRequestSize(maxRequestSize); } if (producerConfiguration.containsKey(SingerConfigDef.SSL_ENABLED_CONFIG)) { boolean enabled = producerConfiguration.getBoolean(SingerConfigDef.SSL_ENABLED_CONFIG); kafkaProducerConfig.setSslEnabled(enabled); if (enabled) { List<String> brokers = kafkaProducerConfig.getBrokerLists(); List<String> updated = new ArrayList<>(); for (int i = 0; i < brokers.size(); i++) { String broker = brokers.get(i); String[] ipports = broker.split(":"); try { InetAddress addr = InetAddress.getByName(ipports[0]); String host = addr.getHostName(); String newBrokerStr = host + ":9093"; updated.add(newBrokerStr); } catch (UnknownHostException e) { LOG.error("Unknown host: {}", ipports[0]); } } kafkaProducerConfig.setBrokerLists(updated); Iterator<String> sslKeysIterator = producerConfiguration.getKeys("ssl"); kafkaProducerConfig.setSslSettings(new HashMap<>()); while (sslKeysIterator.hasNext()) { String key = sslKeysIterator.next(); String value = key.equals(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) ? String.join(",", producerConfiguration.getStringArray(key)) : producerConfiguration.getString(key); kafkaProducerConfig.getSslSettings().put(key, value); } } } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_ENABLED_CONFIG)) { kafkaProducerConfig.setTransactionEnabled(true); } if (producerConfiguration.containsKey(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG)) { int timeoutMs = producerConfiguration.getInt(SingerConfigDef.TRANSACTION_TIMEOUT_MS_CONFIG); kafkaProducerConfig.setTransactionTimeoutMs(timeoutMs); } if (producerConfiguration.containsKey(SingerConfigDef.RETRIES_CONFIG)) { int retries = producerConfiguration.getInt(SingerConfigDef.RETRIES_CONFIG); kafkaProducerConfig.setRetries(retries); } if (producerConfiguration.containsKey(SingerConfigDef.PRODUCER_BUFFER_MEMORY)) { int bufferMemory = producerConfiguration.getInt(SingerConfigDef.PRODUCER_BUFFER_MEMORY); kafkaProducerConfig.setBufferMemory(bufferMemory); } return kafkaProducerConfig; } private  LogConfigUtils(); static SingerConfig parseDirBasedSingerConfigHeader(String singerConfigFile); static MemqWriterConfig parseMemqWriterConfig(SubsetConfiguration configuration); static void parseMemqAuditorConfigs(MemqWriterConfig config,
                                         AbstractConfiguration configuration); static SingerConfig parseFileBasedSingerConfig(String singerConfigFile); static LogMonitorConfig parseLogMonitorConfig(AbstractConfiguration monitorConfiguration); static SingerLogConfig parseLogConfigFromFile(File logConfigFile); static SingerLogConfig[] parseLogStreamConfigFromFile(String logConfigFile); static String parseLogNamespace(File logConfigFile); static SingerLogConfig parseLogConfig(String logName,
                                               AbstractConfiguration logConfiguration); static SingerLogConfig[] parseLogStreamConfig(AbstractConfiguration config); static SingerConfig parseCommonSingerConfigHeader(AbstractConfiguration singerConfiguration); static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration); static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration); static String filePathFromZKPath(String serverSetZKPath); static KafkaProducerConfig parseProducerConfig(AbstractConfiguration producerConfiguration); static PropertiesConfiguration[] getTopicConfigs(Configuration config); static PropertiesConfiguration getTopicConfig(Configuration config, String topicName); static void validateNewConfig(Configuration config); static final String DEFAULT_SHADOW_SERVERSET; static String DEFAULT_SERVERSET_DIR; static final Properties SHADOW_SERVERSET_MAPPING; static boolean SHADOW_MODE_ENABLED; }