DefaultDeduplicator implements AlertDeduplicator { public List<AlertStreamEvent> dedup(AlertStreamEvent event) { if (event == null) { return null; } if (dedupIntervalSec <= 0) { return Collections.singletonList(event); } StreamDefinition streamDefinition = event.getSchema(); HashMap<String, String> customFieldValues = new HashMap<>(); String stateFiledValue = null; for (int i = 0; i < event.getData().length; i++) { if (i > streamDefinition.getColumns().size()) { if (LOG.isWarnEnabled()) { LOG.warn("output column does not found for event data, this indicate code error!"); } continue; } String colName = streamDefinition.getColumns().get(i).getName(); Object colValue = event.getData()[i]; if (colName.equals(dedupStateField) && colValue != null) { stateFiledValue = colValue.toString(); } if (colValue != null) { if (customDedupFields == null || customDedupFields.size() <= 0) { if (streamDefinition.getColumns().get(i).getType().equals(StreamColumn.Type.STRING)) { customFieldValues.put(colName, colValue.toString()); } } else { for (String field : customDedupFields) { if (colName.equals(field)) { customFieldValues.put(field, colValue.toString()); break; } } } } } List<AlertStreamEvent> outputEvents = checkDedup(event, new EventUniq(event.getStreamId(), event.getPolicyId(), event.getCreatedTime(), customFieldValues), stateFiledValue); if (outputEvents != null && outputEvents.size() > 0) { return outputEvents; } else if (LOG.isInfoEnabled()) { LOG.info("Alert event is skipped because it's duplicated: {}", event.toString()); } return null; } DefaultDeduplicator(AlertDeduplication alertDeduplication);  DefaultDeduplicator(String intervalMin, List<String> customDedupFields,
                               String dedupStateField, String dedupStateCloseValue, DedupCache dedupCache); List<AlertStreamEvent> dedup(AlertStreamEvent event); @Override void setDedupIntervalMin(String newDedupIntervalMin); AlertDeduplication getAlertDeduplication();  }
@Test public void testNormal() throws Exception { System.setProperty("config.resource", "/application-mongo-statestore.conf"); Config config = ConfigFactory.load(); DedupCache dedupCache = new DedupCache(config, "testPublishment"); DefaultDeduplicator deduplicator = new DefaultDeduplicator( "PT1M", Arrays.asList(new String[] {"alertKey"}), "state", "close", dedupCache); StreamDefinition stream = createStream(); PolicyDefinition policy = createPolicyGroupByStreamId(stream.getStreamId(), "testPolicy"); AlertStreamEvent e1 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "OPEN", 0, 0 }); AlertStreamEvent e2 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "WARN", 0, 0 }); AlertStreamEvent e3 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "OPEN", 0, 0 }); AlertStreamEvent e4 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "WARN", 0, 0 }); AlertStreamEvent e5 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "CLOSE", 0, 0 }); AlertStreamEvent e6 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "OPEN", 0, 0 }); AlertStreamEvent e7 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "OPEN", 0, 0 }); AlertStreamEvent e8 = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", "OPEN", 0, 0 }); List<AlertStreamEvent> allResults = new ArrayList<AlertStreamEvent>(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e1); if (result != null) { allResults.addAll(result); } System.out.println("1 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e2); if (result != null) { allResults.addAll(result); } System.out.println("2 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e3); if (result != null) { allResults.addAll(result); } System.out.println("3 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e4); if (result != null) { allResults.addAll(result); } System.out.println("4 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { try { Thread.sleep(500); } catch (InterruptedException e) { } List<AlertStreamEvent> result = deduplicator.dedup(e5); if (result != null) { allResults.addAll(result); } System.out.println("5 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e6); if (result != null) { allResults.addAll(result); } System.out.println("6 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e7); if (result != null) { allResults.addAll(result); } System.out.println("7 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); new Thread(new Runnable() { @Override public void run() { List<AlertStreamEvent> result = deduplicator.dedup(e8); if (result != null) { allResults.addAll(result); } System.out.println("8 >>>> " + ToStringBuilder.reflectionToString(result)); } }).start(); Thread.sleep(2000); long maxCount = 0; for (AlertStreamEvent event : allResults) { Assert.assertNotNull(event.getData()[4]); Assert.assertNotNull(event.getData()[5]); if (((Long) event.getData()[4]) > maxCount) { maxCount = (Long) event.getData()[4]; System.out.println(String.format(">>>>>%s: %s", event, maxCount)); } } }