ConfiguredEnrichmentBolt extends ConfiguredBolt<EnrichmentConfigurations> { @Override public void cleanup() { super.cleanup(); try { StellarFunctions.close(); } catch (IOException e) { LOG.error(e.getMessage(), e); } } ConfiguredEnrichmentBolt(String zookeeperUrl); @Override void cleanup();  }
@Test public void test() throws Exception { EnrichmentConfigurations sampleConfigurations = new EnrichmentConfigurations(); UnitTestHelper.setLog4jLevel(ConfiguredBolt.class, Level.FATAL); StandAloneConfiguredEnrichmentBolt configuredBoltNullZk = new StandAloneConfiguredEnrichmentBolt(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(sampleConfigPath)); Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(enrichmentsConfigPath); for (String sensorType : sensorEnrichmentConfigs.keySet()) { sampleConfigurations.updateSensorEnrichmentConfig(sensorType, sensorEnrichmentConfigs.get(sensorType)); } StandAloneConfiguredEnrichmentBolt configuredBolt = new StandAloneConfiguredEnrichmentBolt(zookeeperUrl); configuredBolt.prepare(new HashMap(), topologyContext, outputCollector); waitForConfigUpdate(enrichmentConfigurationTypes); 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"; SensorEnrichmentConfig testSensorConfig = new SensorEnrichmentConfig(); Map<String, Object> enrichmentFieldMap = new HashMap<>(); enrichmentFieldMap.put("enrichmentTest", new ArrayList<String>() {{ add("enrichmentField"); }}); testSensorConfig.getEnrichment().setFieldMap(enrichmentFieldMap); Map<String, Object> threatIntelFieldMap = new HashMap<>(); threatIntelFieldMap.put("threatIntelTest", new ArrayList<String>() {{ add("threatIntelField"); }}); testSensorConfig.getThreatIntel().setFieldMap(threatIntelFieldMap); sampleConfigurations.updateSensorEnrichmentConfig(sensorType, testSensorConfig); ConfigurationsUtils.writeSensorEnrichmentConfigToZookeeper(sensorType, testSensorConfig, zookeeperUrl); waitForConfigUpdate(sensorType); assertEquals(sampleConfigurations, configuredBolt.getConfigurations(), "Add new sensor config"); configuredBolt.cleanup(); }