ConfiguredParserBolt extends ConfiguredBolt<ParserConfigurations> { @Override public void cleanup() { super.cleanup(); try { StellarFunctions.close(); } catch (IOException e) { LOG.error(e.getMessage(), e); } } ConfiguredParserBolt(String zookeeperUrl); @Override void cleanup();  }
@Test public void test() throws Exception { ParserConfigurations sampleConfigurations = new ParserConfigurations(); UnitTestHelper.setLog4jLevel(ConfiguredBolt.class, Level.FATAL); StandAloneConfiguredParserBolt configuredBoltNullZk = new ConfiguredParserBoltTest.StandAloneConfiguredParserBolt(null); assertThrows(RuntimeException.class, () -> configuredBoltNullZk.prepare(new HashMap(), topologyContext, outputCollector), "A valid zookeeper url must be supplied"); UnitTestHelper.setLog4jLevel(ConfiguredBolt.class, Level.ERROR); configsUpdated = new HashSet<>(); sampleConfigurations.updateGlobalConfig(ConfigurationsUtils.readGlobalConfigFromFile("../" + TestConstants.SAMPLE_CONFIG_PATH)); Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile("../" + TestConstants.PARSER_CONFIGS_PATH); for (String sensorType : sensorParserConfigs.keySet()) { sampleConfigurations.updateSensorParserConfig(sensorType, sensorParserConfigs.get(sensorType)); } StandAloneConfiguredParserBolt configuredBolt = new StandAloneConfiguredParserBolt(zookeeperUrl); configuredBolt.prepare(new HashMap(), topologyContext, outputCollector); waitForConfigUpdate(parserConfigurationTypes); assertEquals(sampleConfigurations, configuredBolt.getConfigurations()); configsUpdated = new HashSet<>(); Map<String, Object> sampleGlobalConfig = sampleConfigurations.getGlobalConfig(); sampleGlobalConfig.put("newGlobalField", "newGlobalValue"); ConfigurationsUtils.writeGlobalConfigToZookeeper(sampleGlobalConfig, zookeeperUrl); waitForConfigUpdate(ConfigurationType.GLOBAL.getTypeName()); assertEquals(sampleConfigurations.getGlobalConfig(), configuredBolt.getConfigurations().getGlobalConfig(), "Add global config field"); configsUpdated = new HashSet<>(); sampleGlobalConfig.remove("newGlobalField"); ConfigurationsUtils.writeGlobalConfigToZookeeper(sampleGlobalConfig, zookeeperUrl); waitForConfigUpdate(ConfigurationType.GLOBAL.getTypeName()); assertEquals(sampleConfigurations, configuredBolt.getConfigurations(), "Remove global config field"); configsUpdated = new HashSet<>(); String sensorType = "testSensorConfig"; SensorParserConfig testSensorConfig = new SensorParserConfig(); testSensorConfig.setParserClassName("className"); testSensorConfig.setSensorTopic("sensorTopic"); testSensorConfig.setParserConfig(new HashMap<String, Object>() {{ put("configName", "configObject"); }}); sampleConfigurations.updateSensorParserConfig(sensorType, testSensorConfig); ConfigurationsUtils.writeSensorParserConfigToZookeeper(sensorType, testSensorConfig, zookeeperUrl); waitForConfigUpdate(sensorType); ParserConfigurations configuredBoltConfigs = configuredBolt.getConfigurations(); if(!sampleConfigurations.equals(configuredBoltConfigs)) { if(sampleConfigurations.getFieldValidations().size() != configuredBoltConfigs.getFieldValidations().size()) { System.out.println("Field validations don't line up"); } for(int i = 0;i < sampleConfigurations.getFieldValidations().size();++i) { FieldValidator l = sampleConfigurations.getFieldValidations().get(i); FieldValidator r = configuredBoltConfigs.getFieldValidations().get(i); if(!l.equals(r)) { System.out.println(l + " != " + r); } } if(sampleConfigurations.getConfigurations().size() != configuredBoltConfigs.getConfigurations().size()) { System.out.println("Configs don't line up"); } for(Map.Entry<String, Object> kv : sampleConfigurations.getConfigurations().entrySet() ) { Object l = kv.getValue(); Object r = configuredBoltConfigs.getConfigurations().get(kv.getKey()); if(!l.equals(r)) { System.out.println(kv.getKey() + " config does not line up: " ); System.out.println(l); System.out.println(r); } } assertEquals(sampleConfigurations, configuredBoltConfigs, "Add new sensor config"); } assertEquals(sampleConfigurations, configuredBoltConfigs, "Add new sensor config"); configuredBolt.cleanup(); }