ParserTopologyCLI { private static Optional<String> getSecurityProtocol(Optional<String> protocol, List<Map<String, Object>> spoutConfig) { Optional<String> ret = protocol; if(ret.isPresent() && protocol.get().equalsIgnoreCase(SecurityProtocol.PLAINTEXT.name)) { ret = Optional.empty(); } if(!ret.isPresent()) { String spoutConfigSp = null; for (Map<String, Object> config: spoutConfig) { String configSp = (String) config.get(KafkaUtils.SECURITY_PROTOCOL); if (configSp != null && !SecurityProtocol.PLAINTEXT.name.equals(configSp)) { spoutConfigSp = configSp; } else if (configSp != null) { spoutConfigSp = configSp; } } ret = Optional.ofNullable(spoutConfigSp); } if(ret.isPresent() && ret.get().equalsIgnoreCase(SecurityProtocol.PLAINTEXT.name)) { ret = Optional.empty(); } return ret; }  ParserTopologyBuilder.ParserTopology createParserTopology(final CommandLine cmd); static void main(String[] args); static final String STORM_JOB_SEPARATOR; static final String TOPOLOGY_OPTION_SEPARATOR; }
@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 { 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 { File extraConfig = File.createTempFile("spoutConfig", "json"); extraConfig.deleteOnExit(); writeMap(extraConfig, new HashMap<String, Object>() {{ put("security.protocol", "PLAINTEXTSASL"); }}); { 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") ); } { 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") ); } }