@Test public void testRealPinTTLParsing() throws ConfigurationException { String CONFIG = "" + "topic=test\n" + "objectType=mobile_perf_log\n" + "serverSetPath=/xyz/realpin/prod"; PropertiesConfiguration config = new PropertiesConfiguration(); config.load(new ByteArrayInputStream(CONFIG.getBytes())); RealpinWriterConfig rpConf = LogConfigUtils.parseRealpinWriterConfig(config); assertEquals(-1, rpConf.getTtl()); CONFIG = "" + "topic=test\n" + "objectType=mobile_perf_log\n" + "serverSetPath=/xyz/realpin/prod\n" + "ttl=10000"; config = new PropertiesConfiguration(); config.load(new ByteArrayInputStream(CONFIG.getBytes())); rpConf = LogConfigUtils.parseRealpinWriterConfig(config); assertEquals(10000, rpConf.getTtl()); }
public static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration) { configuration.setThrowExceptionOnMissing(true); String topic = configuration.getString(SingerConfigDef.TOPIC); String objectTypeString = configuration.getString(SingerConfigDef.REALPIN_OBJECT_TYPE); RealpinObjectType objectType = RealpinObjectType.valueOf(objectTypeString.toUpperCase()); String serverSetPath = configuration.getString(SingerConfigDef.REALPIN_SERVERSET_PATH); RealpinWriterConfig writer = new RealpinWriterConfig(topic, objectType, serverSetPath); if (configuration.containsKey(SingerConfigDef.REALPIN_TIMEOUT_MS)) { writer.setTimeoutMs(configuration.getInt(SingerConfigDef.REALPIN_TIMEOUT_MS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_RETRIES)) { writer.setRetries(configuration.getInt(SingerConfigDef.REALPIN_RETRIES)); } if (configuration.containsKey(SingerConfigDef.REALPIN_HOST_LIMIT)) { writer.setHostLimit(configuration.getInt(SingerConfigDef.REALPIN_HOST_LIMIT)); } if (configuration.containsKey(SingerConfigDef.REALPIN_MAX_WAITERS)) { writer.setMaxWaiters(configuration.getInt(SingerConfigDef.REALPIN_MAX_WAITERS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_TTL)) { writer.setTtl(configuration.getInt(SingerConfigDef.REALPIN_TTL)); } return writer; }
LogConfigUtils { public static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration) { configuration.setThrowExceptionOnMissing(true); String topic = configuration.getString(SingerConfigDef.TOPIC); String objectTypeString = configuration.getString(SingerConfigDef.REALPIN_OBJECT_TYPE); RealpinObjectType objectType = RealpinObjectType.valueOf(objectTypeString.toUpperCase()); String serverSetPath = configuration.getString(SingerConfigDef.REALPIN_SERVERSET_PATH); RealpinWriterConfig writer = new RealpinWriterConfig(topic, objectType, serverSetPath); if (configuration.containsKey(SingerConfigDef.REALPIN_TIMEOUT_MS)) { writer.setTimeoutMs(configuration.getInt(SingerConfigDef.REALPIN_TIMEOUT_MS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_RETRIES)) { writer.setRetries(configuration.getInt(SingerConfigDef.REALPIN_RETRIES)); } if (configuration.containsKey(SingerConfigDef.REALPIN_HOST_LIMIT)) { writer.setHostLimit(configuration.getInt(SingerConfigDef.REALPIN_HOST_LIMIT)); } if (configuration.containsKey(SingerConfigDef.REALPIN_MAX_WAITERS)) { writer.setMaxWaiters(configuration.getInt(SingerConfigDef.REALPIN_MAX_WAITERS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_TTL)) { writer.setTtl(configuration.getInt(SingerConfigDef.REALPIN_TTL)); } return writer; } }
LogConfigUtils { public static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration) { configuration.setThrowExceptionOnMissing(true); String topic = configuration.getString(SingerConfigDef.TOPIC); String objectTypeString = configuration.getString(SingerConfigDef.REALPIN_OBJECT_TYPE); RealpinObjectType objectType = RealpinObjectType.valueOf(objectTypeString.toUpperCase()); String serverSetPath = configuration.getString(SingerConfigDef.REALPIN_SERVERSET_PATH); RealpinWriterConfig writer = new RealpinWriterConfig(topic, objectType, serverSetPath); if (configuration.containsKey(SingerConfigDef.REALPIN_TIMEOUT_MS)) { writer.setTimeoutMs(configuration.getInt(SingerConfigDef.REALPIN_TIMEOUT_MS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_RETRIES)) { writer.setRetries(configuration.getInt(SingerConfigDef.REALPIN_RETRIES)); } if (configuration.containsKey(SingerConfigDef.REALPIN_HOST_LIMIT)) { writer.setHostLimit(configuration.getInt(SingerConfigDef.REALPIN_HOST_LIMIT)); } if (configuration.containsKey(SingerConfigDef.REALPIN_MAX_WAITERS)) { writer.setMaxWaiters(configuration.getInt(SingerConfigDef.REALPIN_MAX_WAITERS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_TTL)) { writer.setTtl(configuration.getInt(SingerConfigDef.REALPIN_TTL)); } return writer; } private  LogConfigUtils(); }
LogConfigUtils { public static RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration) { configuration.setThrowExceptionOnMissing(true); String topic = configuration.getString(SingerConfigDef.TOPIC); String objectTypeString = configuration.getString(SingerConfigDef.REALPIN_OBJECT_TYPE); RealpinObjectType objectType = RealpinObjectType.valueOf(objectTypeString.toUpperCase()); String serverSetPath = configuration.getString(SingerConfigDef.REALPIN_SERVERSET_PATH); RealpinWriterConfig writer = new RealpinWriterConfig(topic, objectType, serverSetPath); if (configuration.containsKey(SingerConfigDef.REALPIN_TIMEOUT_MS)) { writer.setTimeoutMs(configuration.getInt(SingerConfigDef.REALPIN_TIMEOUT_MS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_RETRIES)) { writer.setRetries(configuration.getInt(SingerConfigDef.REALPIN_RETRIES)); } if (configuration.containsKey(SingerConfigDef.REALPIN_HOST_LIMIT)) { writer.setHostLimit(configuration.getInt(SingerConfigDef.REALPIN_HOST_LIMIT)); } if (configuration.containsKey(SingerConfigDef.REALPIN_MAX_WAITERS)) { writer.setMaxWaiters(configuration.getInt(SingerConfigDef.REALPIN_MAX_WAITERS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_TTL)) { writer.setTtl(configuration.getInt(SingerConfigDef.REALPIN_TTL)); } return writer; } 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 RealpinWriterConfig parseRealpinWriterConfig(AbstractConfiguration configuration) { configuration.setThrowExceptionOnMissing(true); String topic = configuration.getString(SingerConfigDef.TOPIC); String objectTypeString = configuration.getString(SingerConfigDef.REALPIN_OBJECT_TYPE); RealpinObjectType objectType = RealpinObjectType.valueOf(objectTypeString.toUpperCase()); String serverSetPath = configuration.getString(SingerConfigDef.REALPIN_SERVERSET_PATH); RealpinWriterConfig writer = new RealpinWriterConfig(topic, objectType, serverSetPath); if (configuration.containsKey(SingerConfigDef.REALPIN_TIMEOUT_MS)) { writer.setTimeoutMs(configuration.getInt(SingerConfigDef.REALPIN_TIMEOUT_MS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_RETRIES)) { writer.setRetries(configuration.getInt(SingerConfigDef.REALPIN_RETRIES)); } if (configuration.containsKey(SingerConfigDef.REALPIN_HOST_LIMIT)) { writer.setHostLimit(configuration.getInt(SingerConfigDef.REALPIN_HOST_LIMIT)); } if (configuration.containsKey(SingerConfigDef.REALPIN_MAX_WAITERS)) { writer.setMaxWaiters(configuration.getInt(SingerConfigDef.REALPIN_MAX_WAITERS)); } if (configuration.containsKey(SingerConfigDef.REALPIN_TTL)) { writer.setTtl(configuration.getInt(SingerConfigDef.REALPIN_TTL)); } return writer; } 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; }