| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.metron.parsers.topology; |
| |
| import org.adrianwalker.multilinestring.Multiline; |
| import org.apache.commons.cli.CommandLine; |
| import org.apache.commons.cli.ParseException; |
| import org.apache.commons.cli.Parser; |
| import org.apache.commons.cli.PosixParser; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.log4j.Level; |
| import org.apache.metron.common.configuration.SensorParserConfig; |
| import org.apache.metron.common.utils.JSONUtils; |
| import org.apache.metron.parsers.topology.config.ValueSupplier; |
| import org.apache.metron.test.utils.UnitTestHelper; |
| import org.apache.storm.Config; |
| import org.junit.jupiter.api.Test; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.*; |
| import java.util.function.Predicate; |
| import java.util.function.Supplier; |
| |
| import static org.junit.jupiter.api.Assertions.*; |
| |
| public class ParserTopologyCLITest { |
| |
| |
| public static class CLIBuilder { |
| EnumMap<ParserTopologyCLI.ParserOptions, String> map = new EnumMap<>(ParserTopologyCLI.ParserOptions.class); |
| |
| public CLIBuilder with(ParserTopologyCLI.ParserOptions option, String val) { |
| map.put(option, val); |
| return this; |
| } |
| public CLIBuilder with(ParserTopologyCLI.ParserOptions option) { |
| map.put(option, null); |
| return this; |
| } |
| public CommandLine build(boolean longOpt) throws ParseException { |
| return getCLI(map, longOpt); |
| } |
| private CommandLine getCLI(EnumMap<ParserTopologyCLI.ParserOptions, String> options, boolean longOpt) throws ParseException { |
| ArrayList<String> args = new ArrayList<>(); |
| for (Map.Entry<ParserTopologyCLI.ParserOptions, String> option : options.entrySet()) { |
| boolean hasLongOpt = option.getKey().option.hasLongOpt(); |
| if (hasLongOpt && longOpt) { |
| args.add("--" + option.getKey().option.getLongOpt()); |
| if (option.getKey().option.hasArg() && option.getValue() != null) { |
| args.add(option.getValue()); |
| } |
| } else if (hasLongOpt && !longOpt) { |
| args.add("-" + option.getKey().shortCode); |
| if (option.getKey().option.hasArg() && option.getValue() != null) { |
| args.add(option.getValue()); |
| } |
| } |
| } |
| return ParserTopologyCLI.ParserOptions.parse(new PosixParser(), args.toArray(new String[args.size()])); |
| } |
| } |
| |
| @Test |
| public void testNoOverlappingArgs() throws Exception { |
| Set<String> optionStrs = new HashSet<>(); |
| for(ParserTopologyCLI.ParserOptions option : ParserTopologyCLI.ParserOptions.values()) { |
| if(optionStrs.contains(option.option.getLongOpt())) { |
| throw new IllegalStateException("Reused long option: " + option.option.getLongOpt()); |
| } |
| if(optionStrs.contains(option.shortCode)) { |
| throw new IllegalStateException("Reused short option: " + option.shortCode); |
| } |
| optionStrs.add(option.option.getLongOpt()); |
| optionStrs.add(option.shortCode); |
| } |
| } |
| |
| @Test |
| public void testKafkaOffset_happyPath() throws ParseException { |
| kafkaOffset(true); |
| kafkaOffset(false); |
| } |
| public void kafkaOffset(boolean longOpt) throws ParseException { |
| CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor") |
| .build(longOpt); |
| assertEquals("myzk", ParserTopologyCLI.ParserOptions.ZK_QUORUM.get(cli)); |
| assertEquals("mybroker", ParserTopologyCLI.ParserOptions.BROKER_URL.get(cli)); |
| assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPES.get(cli)); |
| } |
| @Test |
| public void testCLI_happyPath() throws ParseException { |
| happyPath(true); |
| happyPath(false); |
| } |
| |
| @Test |
| public void testCLI_insufficientArg() { |
| UnitTestHelper.setLog4jLevel(Parser.class, Level.FATAL); |
| assertThrows(ParseException.class, () -> |
| new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .build(true)); |
| UnitTestHelper.setLog4jLevel(Parser.class, Level.ERROR); |
| } |
| |
| public void happyPath(boolean longOpt) throws ParseException { |
| CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor") |
| .build(longOpt); |
| assertEquals("myzk", ParserTopologyCLI.ParserOptions.ZK_QUORUM.get(cli)); |
| assertEquals("mybroker", ParserTopologyCLI.ParserOptions.BROKER_URL.get(cli)); |
| assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPES.get(cli)); |
| } |
| |
| @Test |
| public void testConfig_noExtra() throws ParseException { |
| testConfig_noExtra(true); |
| testConfig_noExtra(false); |
| } |
| |
| public void testConfig_noExtra(boolean longOpt) throws ParseException { |
| CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor") |
| .with(ParserTopologyCLI.ParserOptions.NUM_WORKERS, "1") |
| .with(ParserTopologyCLI.ParserOptions.NUM_ACKERS, "2") |
| .with(ParserTopologyCLI.ParserOptions.NUM_MAX_TASK_PARALLELISM, "3") |
| .with(ParserTopologyCLI.ParserOptions.MESSAGE_TIMEOUT, "4") |
| .build(longOpt); |
| Optional<Config> configOptional = ParserTopologyCLI.ParserOptions.getConfig(cli); |
| Config config = configOptional.get(); |
| assertEquals(1, config.get(Config.TOPOLOGY_WORKERS)); |
| assertEquals(2, config.get(Config.TOPOLOGY_ACKER_EXECUTORS)); |
| assertEquals(3, config.get(Config.TOPOLOGY_MAX_TASK_PARALLELISM)); |
| assertEquals(4, config.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)); |
| } |
| |
| @Test |
| public void testOutputTopic() throws Exception { |
| testOutputTopic(true); |
| testOutputTopic(false); |
| } |
| |
| public void testOutputTopic(boolean longOpt) throws ParseException { |
| CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor") |
| .with(ParserTopologyCLI.ParserOptions.OUTPUT_TOPIC, "my_topic") |
| .build(longOpt); |
| assertEquals("my_topic", ParserTopologyCLI.ParserOptions.OUTPUT_TOPIC.get(cli)); |
| } |
| |
| /** |
| { |
| "string" : "foo" |
| ,"integer" : 1 |
| } |
| */ |
| @Multiline |
| public static String extraConfig; |
| |
| @Test |
| public void testConfig_extra() throws Exception { |
| testConfig_extra(true); |
| testConfig_extra(false); |
| } |
| |
| public void testConfig_extra(boolean longOpt) throws IOException, ParseException { |
| File extraFile = File.createTempFile("extra", "json"); |
| try { |
| FileUtils.write(extraFile, extraConfig); |
| CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor") |
| .with(ParserTopologyCLI.ParserOptions.MESSAGE_TIMEOUT, "4") |
| .with(ParserTopologyCLI.ParserOptions.EXTRA_OPTIONS, extraFile.getAbsolutePath()) |
| .build(longOpt); |
| Optional<Config> configOptional = ParserTopologyCLI.ParserOptions.getConfig(cli); |
| Config config = configOptional.get(); |
| assertEquals(4, config.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)); |
| assertEquals("foo", config.get("string")); |
| assertEquals(1, config.get("integer")); |
| } finally{ |
| extraFile.deleteOnExit(); |
| } |
| } |
| |
| private static class ParserInput { |
| private List<Integer> spoutParallelism; |
| private List<Integer> spoutNumTasks; |
| private Integer parserParallelism; |
| private Integer parserNumTasks; |
| private Integer errorParallelism; |
| private Integer errorNumTasks; |
| private List<Map<String, Object>> spoutConfig; |
| private String securityProtocol; |
| private Config stormConf; |
| private String outputTopic; |
| private String errorTopic; |
| |
| public ParserInput(ValueSupplier<List> spoutParallelism, |
| ValueSupplier<List> spoutNumTasks, |
| ValueSupplier<Integer> parserParallelism, |
| ValueSupplier<Integer> parserNumTasks, |
| ValueSupplier<Integer> errorParallelism, |
| ValueSupplier<Integer> errorNumTasks, |
| ValueSupplier<List> spoutConfig, |
| ValueSupplier<String> securityProtocol, |
| ValueSupplier<Config> stormConf, |
| ValueSupplier<String> outputTopic, |
| ValueSupplier<String> errorTopic, |
| List<SensorParserConfig> configs |
| ) |
| { |
| this.spoutParallelism = spoutParallelism.get(configs, List.class); |
| this.spoutNumTasks = spoutNumTasks.get(configs, List.class); |
| this.parserParallelism = parserParallelism.get(configs, Integer.class); |
| this.parserNumTasks = parserNumTasks.get(configs, Integer.class); |
| this.errorParallelism = errorParallelism.get(configs, Integer.class); |
| this.errorNumTasks = errorNumTasks.get(configs, Integer.class); |
| this.spoutConfig = spoutConfig.get(configs, List.class); |
| this.securityProtocol = securityProtocol.get(configs, String.class); |
| this.stormConf = stormConf.get(configs, Config.class); |
| this.outputTopic = outputTopic.get(configs, String.class); |
| this.errorTopic = errorTopic.get(configs, String.class); |
| } |
| |
| public List<Integer> getSpoutParallelism() { |
| return spoutParallelism; |
| } |
| |
| public List<Integer> getSpoutNumTasks() { |
| return spoutNumTasks; |
| } |
| |
| public Integer getParserParallelism() { |
| return parserParallelism; |
| } |
| |
| public Integer getParserNumTasks() { |
| return parserNumTasks; |
| } |
| |
| public Integer getErrorParallelism() { |
| return errorParallelism; |
| } |
| |
| public Integer getErrorNumTasks() { |
| return errorNumTasks; |
| } |
| |
| public List<Map<String, Object>> getSpoutConfig() { |
| return spoutConfig; |
| } |
| |
| public String getSecurityProtocol() { |
| return securityProtocol; |
| } |
| |
| public Config getStormConf() { |
| return stormConf; |
| } |
| |
| public String getOutputTopic() { |
| return outputTopic; |
| } |
| |
| public String getErrorTopic() { |
| return errorTopic; |
| } |
| } |
| |
| /** |
| * { |
| * "parserClassName": "org.apache.metron.parsers.GrokParser", |
| * "sensorTopic": "squid", |
| * "parserConfig": { |
| * "grokPath": "/patterns/squid", |
| * "patternLabel": "SQUID_DELIMITED", |
| * "timestampField": "timestamp" |
| * }, |
| * "fieldTransformations" : [ |
| * { |
| * "transformation" : "STELLAR", |
| * "output" : [ |
| * "full_hostname", |
| * "domain_without_subdomains" |
| * ], |
| * "config" : { |
| * "full_hostname" : "URL_TO_HOST(url)", |
| * "domain_without_subdomains" : "DOMAIN_REMOVE_SUBDOMAINS(full_hostname)" |
| * } |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| public static String baseConfig; |
| |
| private static SensorParserConfig getBaseConfig() { |
| try { |
| return JSONUtils.INSTANCE.load(baseConfig, SensorParserConfig.class); |
| } catch (IOException e) { |
| throw new IllegalStateException(e.getMessage(), e); |
| } |
| } |
| |
| @Test |
| public void testSpoutParallelism() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_PARALLELISM |
| , "10" |
| , input -> input.getSpoutParallelism().equals(Collections.singletonList(10)) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSpoutParallelism(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getSpoutParallelism().equals(Collections.singletonList(20)) |
| ); |
| } |
| |
| @Test |
| public void testSpoutParallelismMultiple() throws Exception { |
| // Each spout uses it's own |
| // Return one per spout. |
| List<Integer> spoutParCli = new ArrayList<>(); |
| spoutParCli.add(10); |
| spoutParCli.add(12); |
| List<Integer> spoutParConfig = new ArrayList<>(); |
| spoutParConfig.add(20); |
| spoutParConfig.add(30); |
| testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_PARALLELISM |
| , "10,12" |
| , input -> input.getSpoutParallelism().equals(spoutParCli) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSpoutParallelism(20); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setSpoutParallelism(30); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| return configs; |
| } |
| , input -> input.getSpoutParallelism().equals(spoutParConfig) |
| ); |
| } |
| |
| @Test |
| public void testSpoutNumTasks() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_NUM_TASKS |
| , "10" |
| , input -> input.getSpoutNumTasks().equals(Collections.singletonList(10)) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSpoutNumTasks(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getSpoutNumTasks().equals(Collections.singletonList(20)) |
| ); |
| } |
| |
| @Test |
| public void testSpoutNumTasksMultiple() throws Exception { |
| // Return one per spout. |
| List<Integer> numTasksCli = new ArrayList<>(); |
| numTasksCli.add(10); |
| numTasksCli.add(12); |
| List<Integer> numTasksConfig = new ArrayList<>(); |
| numTasksConfig.add(20); |
| numTasksConfig.add(30); |
| testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_NUM_TASKS |
| , "10,12" |
| , input -> input.getSpoutNumTasks().equals(numTasksCli) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSpoutNumTasks(20); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setSpoutNumTasks(30); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| return configs; |
| } |
| , input -> input.getSpoutNumTasks().equals(numTasksConfig) |
| ); |
| } |
| |
| @Test |
| public void testParserParallelism() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_PARALLELISM |
| , "10" |
| , input -> input.getParserParallelism().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setParserParallelism(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getParserParallelism().equals(20) |
| ); |
| } |
| |
| @Test |
| public void testParserParallelismMultiple() throws Exception { |
| // Last one wins |
| testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_PARALLELISM |
| , "10" |
| , input -> input.getParserParallelism().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setParserParallelism(20); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setParserParallelism(30); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| return configs; |
| } |
| , input -> input.getParserParallelism().equals(30) |
| ); |
| } |
| |
| @Test |
| public void testParserNumTasks() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_NUM_TASKS |
| , "10" |
| , input -> input.getParserNumTasks().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setParserNumTasks(20); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setParserNumTasks(30); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| return configs; |
| } |
| , input -> input.getParserNumTasks().equals(30) |
| ); |
| } |
| |
| @Test |
| public void testParserNumTasksMultiple() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_NUM_TASKS |
| , "10" |
| , input -> input.getParserNumTasks().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setParserNumTasks(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getParserNumTasks().equals(20) |
| ); |
| } |
| |
| @Test |
| public void testErrorParallelism() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.ERROR_WRITER_PARALLELISM |
| , "10" |
| , input -> input.getErrorParallelism().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setErrorWriterParallelism(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getErrorParallelism().equals(20) |
| ); |
| } |
| |
| @Test |
| public void testErrorNumTasks() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.ERROR_WRITER_NUM_TASKS |
| , "10" |
| , input -> input.getErrorNumTasks().equals(10) |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setErrorWriterNumTasks(20); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getErrorNumTasks().equals(20) |
| ); |
| } |
| |
| @Test |
| public void testSecurityProtocol_fromCLI() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL |
| , "PLAINTEXT" |
| , input -> input.getSecurityProtocol().equals("PLAINTEXT") |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSecurityProtocol("KERBEROS"); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getSecurityProtocol().equals("KERBEROS") |
| ); |
| } |
| |
| @Test |
| public void testSecurityProtocol_fromCLIMultipleUniform() throws Exception { |
| testConfigOption(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL |
| , "PLAINTEXT" |
| , input -> input.getSecurityProtocol().equals("PLAINTEXT") |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSecurityProtocol("PLAINTEXT"); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setSecurityProtocol("PLAINTEXT"); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| return configs; |
| } |
| , input -> input.getSecurityProtocol().equals("PLAINTEXT") |
| ); |
| } |
| |
| @Test |
| public void testSecurityProtocol_fromCLIMultipleMixed() throws Exception { |
| // Non plaintext wins |
| testConfigOption(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL |
| , "PLAINTEXT" |
| , input -> input.getSecurityProtocol().equals("PLAINTEXT") |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSecurityProtocol("PLAINTEXT"); |
| SensorParserConfig config2 = getBaseConfig(); |
| config2.setSecurityProtocol("KERBEROS"); |
| SensorParserConfig config3 = getBaseConfig(); |
| config3.setSecurityProtocol("PLAINTEXT"); |
| List<SensorParserConfig> configs = new ArrayList<>(); |
| configs.add(config); |
| configs.add(config2); |
| configs.add(config3); |
| return configs; |
| } |
| , input -> input.getSecurityProtocol().equals("KERBEROS") |
| ); |
| } |
| |
| @Test |
| public void testSecurityProtocol_fromSpout() throws Exception { |
| //Ultimately the order of precedence is CLI > spout config > parser config |
| File extraConfig = File.createTempFile("spoutConfig", "json"); |
| extraConfig.deleteOnExit(); |
| writeMap(extraConfig, new HashMap<String, Object>() {{ |
| put("security.protocol", "PLAINTEXTSASL"); |
| }}); |
| { |
| //Ensure that the CLI spout config takes precedence |
| |
| testConfigOption(new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) {{ |
| put(ParserTopologyCLI.ParserOptions.SPOUT_CONFIG, extraConfig.getAbsolutePath()); |
| put(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL, "PLAINTEXT"); |
| }} |
| , input -> input.getSecurityProtocol().equals("PLAINTEXT") |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSecurityProtocol("PLAINTEXTSASL_FROM_ZK"); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getSecurityProtocol().equals("PLAINTEXTSASL_FROM_ZK") |
| ); |
| } |
| { |
| //Ensure that the spout config takes precedence |
| testConfigOption(new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) {{ |
| put(ParserTopologyCLI.ParserOptions.SPOUT_CONFIG, extraConfig.getAbsolutePath()); |
| }} |
| , input -> input.getSecurityProtocol().equals("PLAINTEXTSASL") |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSecurityProtocol("PLAINTEXTSASL_FROM_ZK"); |
| return Collections.singletonList(config); |
| } |
| , input -> input.getSecurityProtocol().equals("PLAINTEXTSASL_FROM_ZK") |
| ); |
| } |
| } |
| |
| @Test |
| public void testTopologyConfig_fromConfigExplicitly() throws Exception { |
| testConfigOption(new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) |
| {{ |
| put(ParserTopologyCLI.ParserOptions.NUM_WORKERS, "10"); |
| put(ParserTopologyCLI.ParserOptions.NUM_ACKERS, "20"); |
| }} |
| , input -> { |
| Config c = input.getStormConf(); |
| return (int)c.get(Config.TOPOLOGY_WORKERS) == 10 |
| && (int)c.get(Config.TOPOLOGY_ACKER_EXECUTORS) == 20; |
| } |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setNumWorkers(100); |
| config.setNumAckers(200); |
| return Collections.singletonList(config); |
| } |
| , input -> { |
| Config c = input.getStormConf(); |
| return (int)c.get(Config.TOPOLOGY_WORKERS) == 100 |
| && (int)c.get(Config.TOPOLOGY_ACKER_EXECUTORS) == 200 |
| ; |
| } |
| ); |
| } |
| |
| @Test |
| public void testTopologyConfig() throws Exception { |
| File extraConfig = File.createTempFile("topologyConfig", "json"); |
| extraConfig.deleteOnExit(); |
| writeMap(extraConfig, new HashMap<String, Object>() {{ |
| put(Config.TOPOLOGY_DEBUG, true); |
| }}); |
| testConfigOption(new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) |
| {{ |
| put(ParserTopologyCLI.ParserOptions.NUM_WORKERS, "10"); |
| put(ParserTopologyCLI.ParserOptions.NUM_ACKERS, "20"); |
| put(ParserTopologyCLI.ParserOptions.EXTRA_OPTIONS, extraConfig.getAbsolutePath()); |
| }} |
| , input -> { |
| Config c = input.getStormConf(); |
| return (int)c.get(Config.TOPOLOGY_WORKERS) == 10 |
| && (int)c.get(Config.TOPOLOGY_ACKER_EXECUTORS) == 20 |
| && (boolean)c.get(Config.TOPOLOGY_DEBUG); |
| } |
| , () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setStormConfig( |
| new HashMap<String, Object>() {{ |
| put(Config.TOPOLOGY_WORKERS, 100); |
| put(Config.TOPOLOGY_ACKER_EXECUTORS, 200); |
| }} |
| ); |
| return Collections.singletonList(config); |
| } |
| , input -> { |
| Config c = input.getStormConf(); |
| return (int)c.get(Config.TOPOLOGY_WORKERS) == 100 |
| && (int)c.get(Config.TOPOLOGY_ACKER_EXECUTORS) == 200 |
| && !c.containsKey(Config.TOPOLOGY_DEBUG); |
| } |
| ); |
| } |
| |
| @Test |
| public void testSpoutConfig() throws Exception { |
| File extraConfig = File.createTempFile("spoutConfig", "json"); |
| extraConfig.deleteOnExit(); |
| writeMap(extraConfig, new HashMap<String, Object>() {{ |
| put("extra_config", "from_file"); |
| }}); |
| EnumMap<ParserTopologyCLI.ParserOptions, String> cliOptions = new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) |
| {{ |
| put(ParserTopologyCLI.ParserOptions.SPOUT_CONFIG, extraConfig.getAbsolutePath()); |
| }}; |
| Predicate<ParserInput> cliOverrideExpected = input -> { |
| return input.getSpoutConfig().get(0).get("extra_config").equals("from_file"); |
| }; |
| |
| Predicate<ParserInput> configOverrideExpected = input -> { |
| return input.getSpoutConfig().get(0).get("extra_config").equals("from_zk"); |
| }; |
| |
| Supplier<List<SensorParserConfig>> configSupplier = () -> { |
| SensorParserConfig config = getBaseConfig(); |
| config.setSpoutConfig( |
| new HashMap<String, Object>() {{ |
| put("extra_config", "from_zk"); |
| }} |
| ); |
| return Collections.singletonList(config); |
| }; |
| testConfigOption( cliOptions |
| , cliOverrideExpected |
| , configSupplier |
| , configOverrideExpected |
| ); |
| } |
| |
| private void writeMap(File outFile, Map<String, Object> config) throws IOException { |
| FileUtils.write(outFile, JSONUtils.INSTANCE.toJSON(config, true)); |
| } |
| |
| private void testConfigOption( ParserTopologyCLI.ParserOptions option |
| , String cliOverride |
| , Predicate<ParserInput> cliOverrideCondition |
| , Supplier<List<SensorParserConfig>> configSupplier |
| , Predicate<ParserInput> configOverrideCondition |
| ) throws Exception { |
| testConfigOption( |
| new EnumMap<ParserTopologyCLI.ParserOptions, String>(ParserTopologyCLI.ParserOptions.class) {{ |
| put(option, cliOverride); |
| }}, |
| cliOverrideCondition, |
| configSupplier, |
| configOverrideCondition |
| ); |
| } |
| |
| private void testConfigOption( EnumMap<ParserTopologyCLI.ParserOptions, String> options |
| , Predicate<ParserInput> cliOverrideCondition |
| , Supplier<List<SensorParserConfig>> configSupplier |
| , Predicate<ParserInput> configOverrideCondition |
| ) throws Exception { |
| //CLI Override |
| List<SensorParserConfig> configs = configSupplier.get(); |
| { |
| CLIBuilder builder = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor"); |
| for(Map.Entry<ParserTopologyCLI.ParserOptions, String> entry : options.entrySet()) { |
| builder.with(entry.getKey(), entry.getValue()); |
| } |
| CommandLine cmd = builder.build(true); |
| ParserInput input = getInput(cmd, configs); |
| assertTrue(cliOverrideCondition.test(input)); |
| } |
| // Config Override |
| { |
| CLIBuilder builder = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker") |
| .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk") |
| .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor"); |
| CommandLine cmd = builder.build(true); |
| ParserInput input = getInput(cmd, configs); |
| assertTrue(configOverrideCondition.test(input)); |
| } |
| } |
| |
| private static ParserInput getInput(CommandLine cmd, List<SensorParserConfig> configs ) throws Exception { |
| final ParserInput[] parserInput = new ParserInput[]{null}; |
| new ParserTopologyCLI() { |
| @Override |
| protected ParserTopologyBuilder.ParserTopology getParserTopology( |
| String zookeeperUrl, |
| Optional<String> brokerUrl, |
| List<String> sensorType, |
| ValueSupplier<List> spoutParallelism, |
| ValueSupplier<List> spoutNumTasks, |
| ValueSupplier<Integer> parserParallelism, |
| ValueSupplier<Integer> parserNumTasks, |
| ValueSupplier<Integer> errorParallelism, |
| ValueSupplier<Integer> errorNumTasks, |
| ValueSupplier<List> spoutConfig, |
| ValueSupplier<String> securityProtocol, |
| ValueSupplier<Config> stormConf, |
| ValueSupplier<String> outputTopic, |
| ValueSupplier<String> errorTopic) throws Exception { |
| |
| parserInput[0] = new ParserInput( |
| spoutParallelism, |
| spoutNumTasks, |
| parserParallelism, |
| parserNumTasks, |
| errorParallelism, |
| errorNumTasks, |
| spoutConfig, |
| securityProtocol, |
| stormConf, |
| outputTopic, |
| errorTopic, |
| configs |
| ); |
| |
| return null; |
| } |
| }.createParserTopology(cmd); |
| return parserInput[0]; |
| } |
| |
| } |
