SimpleHBaseAdapter implements EnrichmentAdapter<CacheKey>,Serializable { @Override public JSONObject enrich(CacheKey value) { JSONObject enriched = new JSONObject(); if(!isInitialized()) { initializeAdapter(null); } List<String> enrichmentTypes = value.getConfig() .getEnrichment().getFieldToTypeMap() .get(EnrichmentUtils.toTopLevelField(value.getField())); if(isInitialized() && enrichmentTypes != null && value.getValue() != null) { try { for (LookupKV<EnrichmentKey, EnrichmentValue> kv : lookup.get(Iterables.transform(enrichmentTypes , new EnrichmentUtils.TypeToKey( value.coerceValue(String.class) , lookup.getTable() , value.getConfig().getEnrichment() ) ) , false ) ) { if (kv != null && kv.getValue() != null && kv.getValue().getMetadata() != null) { for (Map.Entry<String, Object> values : kv.getValue().getMetadata().entrySet()) { enriched.put(kv.getKey().type + "." + values.getKey(), values.getValue()); } LOG.trace("Enriched type {} => {}", () -> kv.getKey().type, ()->enriched); } } } catch (IOException e) { LOG.error("Unable to retrieve value: {}", e.getMessage(), e); initializeAdapter(null); throw new RuntimeException("Unable to retrieve value: " + e.getMessage(), e); } } LOG.trace("SimpleHBaseAdapter succeeded: {}", enriched); return enriched; } SimpleHBaseAdapter();  SimpleHBaseAdapter(SimpleHBaseConfig config); SimpleHBaseAdapter withConfig(SimpleHBaseConfig config); @Override void logAccess(CacheKey value); boolean isInitialized(); @Override JSONObject enrich(CacheKey value); @Override boolean initializeAdapter(Map<String, Object> configuration); @Override void updateAdapter(Map<String, Object> config); @Override void cleanup(); @Override String getOutputPrefix(CacheKey value);  }
@Test public void testEnrich() throws Exception { SimpleHBaseAdapter sha = new SimpleHBaseAdapter(); sha.lookup = lookup; SensorEnrichmentConfig broSc = JSONUtils.INSTANCE.load(sourceConfigStr, SensorEnrichmentConfig.class); JSONObject actualMessage = sha.enrich(new CacheKey("test", "test", broSc)); assertEquals(actualMessage, new JSONObject()); actualMessage = sha.enrich(new CacheKey("ip_dst_addr", "10.0.2.3", broSc)); assertNotNull(actualMessage); assertEquals(expectedMessage, actualMessage); }
@Test public void testEnrichNonStringValue() throws Exception { SimpleHBaseAdapter sha = new SimpleHBaseAdapter(); sha.lookup = lookup; SensorEnrichmentConfig broSc = JSONUtils.INSTANCE.load(sourceConfigStr, SensorEnrichmentConfig.class); JSONObject actualMessage = sha.enrich(new CacheKey("test", "test", broSc)); assertEquals(actualMessage, new JSONObject()); actualMessage = sha.enrich(new CacheKey("ip_dst_addr", 10L, broSc)); assertEquals(actualMessage,new JSONObject()); }
@Test public void testMultiColumnFamilies() throws Exception { SimpleHBaseAdapter sha = new SimpleHBaseAdapter(); sha.lookup = lookup; SensorEnrichmentConfig broSc = JSONUtils.INSTANCE.load(sourceConfigWithCFStr, SensorEnrichmentConfig.class); JSONObject actualMessage = sha.enrich(new CacheKey("test", "test", broSc)); assertEquals(actualMessage, new JSONObject()); actualMessage = sha.enrich(new CacheKey("ip_dst_addr", "10.0.2.4", broSc)); assertNotNull(actualMessage); assertEquals(new JSONObject(ImmutableMap.of("cf1.key", "value")), actualMessage); }
@Test public void testMultiColumnFamiliesWrongCF() throws Exception { SimpleHBaseAdapter sha = new SimpleHBaseAdapter(); sha.lookup = lookup; SensorEnrichmentConfig broSc = JSONUtils.INSTANCE.load(sourceConfigStr, SensorEnrichmentConfig.class); JSONObject actualMessage = sha.enrich(new CacheKey("test", "test", broSc)); assertEquals(actualMessage, new JSONObject()); actualMessage = sha.enrich(new CacheKey("ip_dst_addr", "10.0.2.4", broSc)); assertNotNull(actualMessage); assertEquals(new JSONObject(new HashMap<String, Object>()), actualMessage); }