@Test public void testMemqConfigurations() throws Exception { LogConfigUtils.DEFAULT_SERVERSET_DIR = "target"; String config = "type=memq\n" + "memq.topic=test2\n" + "memq.cluster=prototype\n" + "memq.environment=dev\n" + "memq.compression=zstd\n" + "memq.maxInFlightRequests=60\n" + "memq.disableAcks=false\n" + "memq.maxPayLoadBytes=2010000\n" + "memq.clientType=tcp\n" + "memq.auditor.enabled=true\n" + "memq.auditor.topic=auditTopic\n" + "memq.auditor.class=com.pinterest.memq.client.commons.audit.KafkaBackedAuditor\n" + "memq.auditor.serverset=/var/serverset/discovery.testkafka.prod"; PropertiesConfiguration conf = new PropertiesConfiguration(); conf.load(new ByteArrayInputStream(config.getBytes())); String pathname = "target/discovery.memq.dev.prototype.prod_rich_data"; File file = new File(pathname); Path path = file.toPath(); if (file.exists()) { Files.delete(path); } Files.write(path, "test".getBytes()); MemqWriterConfig writerConfig = LogConfigUtils.parseLogStreamWriterConfig(conf) .getMemqWriterConfig(); assertNotNull(writerConfig); assertEquals("prototype", writerConfig.getCluster()); assertEquals("test2", writerConfig.getTopic()); assertNotNull(writerConfig.getAuditorConfig()); assertEquals("target/discovery.memq.dev.prototype.prod_rich_data", writerConfig.getServerset()); }
public static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration) throws ConfigurationException { writerConfiguration.setThrowExceptionOnMissing(true); String writerTypeString = writerConfiguration.getString("type"); WriterType type = WriterType.valueOf(writerTypeString.toUpperCase()); LogStreamWriterConfig writerConfig = new LogStreamWriterConfig(type); switch (type) { case KAFKA08: case KAFKA: writerConfig.setKafkaWriterConfig(parseKafkaWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case REALPIN: writerConfig.setRealpinWriterConfig(parseRealpinWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case NO_OP: writerConfig.setNoOpWriteConfig(parseNoOpWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case PULSAR: writerConfig.setPulsarWriterConfig(parsePulsarWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case MEMQ: writerConfig.setMemqWriterConfig(parseMemqWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; default: throw new ConfigurationException("Unsupported log writer type."); } }
LogConfigUtils { public static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration) throws ConfigurationException { writerConfiguration.setThrowExceptionOnMissing(true); String writerTypeString = writerConfiguration.getString("type"); WriterType type = WriterType.valueOf(writerTypeString.toUpperCase()); LogStreamWriterConfig writerConfig = new LogStreamWriterConfig(type); switch (type) { case KAFKA08: case KAFKA: writerConfig.setKafkaWriterConfig(parseKafkaWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case REALPIN: writerConfig.setRealpinWriterConfig(parseRealpinWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case NO_OP: writerConfig.setNoOpWriteConfig(parseNoOpWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case PULSAR: writerConfig.setPulsarWriterConfig(parsePulsarWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case MEMQ: writerConfig.setMemqWriterConfig(parseMemqWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; default: throw new ConfigurationException("Unsupported log writer type."); } } }
LogConfigUtils { public static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration) throws ConfigurationException { writerConfiguration.setThrowExceptionOnMissing(true); String writerTypeString = writerConfiguration.getString("type"); WriterType type = WriterType.valueOf(writerTypeString.toUpperCase()); LogStreamWriterConfig writerConfig = new LogStreamWriterConfig(type); switch (type) { case KAFKA08: case KAFKA: writerConfig.setKafkaWriterConfig(parseKafkaWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case REALPIN: writerConfig.setRealpinWriterConfig(parseRealpinWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case NO_OP: writerConfig.setNoOpWriteConfig(parseNoOpWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case PULSAR: writerConfig.setPulsarWriterConfig(parsePulsarWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case MEMQ: writerConfig.setMemqWriterConfig(parseMemqWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; default: throw new ConfigurationException("Unsupported log writer type."); } } private  LogConfigUtils(); }
LogConfigUtils { public static LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration) throws ConfigurationException { writerConfiguration.setThrowExceptionOnMissing(true); String writerTypeString = writerConfiguration.getString("type"); WriterType type = WriterType.valueOf(writerTypeString.toUpperCase()); LogStreamWriterConfig writerConfig = new LogStreamWriterConfig(type); switch (type) { case KAFKA08: case KAFKA: writerConfig.setKafkaWriterConfig(parseKafkaWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case REALPIN: writerConfig.setRealpinWriterConfig(parseRealpinWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case NO_OP: writerConfig.setNoOpWriteConfig(parseNoOpWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case PULSAR: writerConfig.setPulsarWriterConfig(parsePulsarWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case MEMQ: writerConfig.setMemqWriterConfig(parseMemqWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; default: throw new ConfigurationException("Unsupported log writer type."); } } 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 LogStreamWriterConfig parseLogStreamWriterConfig(AbstractConfiguration writerConfiguration) throws ConfigurationException { writerConfiguration.setThrowExceptionOnMissing(true); String writerTypeString = writerConfiguration.getString("type"); WriterType type = WriterType.valueOf(writerTypeString.toUpperCase()); LogStreamWriterConfig writerConfig = new LogStreamWriterConfig(type); switch (type) { case KAFKA08: case KAFKA: writerConfig.setKafkaWriterConfig(parseKafkaWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case REALPIN: writerConfig.setRealpinWriterConfig(parseRealpinWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case NO_OP: writerConfig.setNoOpWriteConfig(parseNoOpWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case PULSAR: writerConfig.setPulsarWriterConfig(parsePulsarWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; case MEMQ: writerConfig.setMemqWriterConfig(parseMemqWriterConfig( new SubsetConfiguration(writerConfiguration, writerTypeString + "."))); return writerConfig; default: throw new ConfigurationException("Unsupported log writer type."); } } 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; }