@Test public void testApplyWithFilterOutByDedupLast() { EasyMock.expect(inputMock.getCache()).andReturn(createLruCache(DEFAULT_DUMMY_MESSAGE, 10L, true)); EasyMock.expect(inputMock.getConfigs()).andReturn(generateInputConfigs()); EasyMock.expect(inputMock.getCacheKeyField()).andReturn(CACHE_KEY_FIELD); EasyMock.replay(inputMock); boolean result = underTest.apply(generateLineMap(), inputMock); EasyMock.verify(inputMock); assertTrue(result); }
public Boolean apply(Map<String, Object> lineMap, Input input) { boolean isLogFilteredOut = false; LRUCache inputLruCache = input.getCache(); if (inputLruCache != null && "service".equals(input.getConfigs().get(LogFeederConstants.ROW_TYPE))) { String logMessage = (String) lineMap.get(input.getCacheKeyField()); Long timestamp = null; if (lineMap.containsKey((LogFeederConstants.IN_MEMORY_TIMESTAMP))) { timestamp = (Long) lineMap.get(LogFeederConstants.IN_MEMORY_TIMESTAMP); } if (logMessage != null && timestamp != null) { isLogFilteredOut = !inputLruCache.isEntryReplaceable(logMessage, timestamp); if (!isLogFilteredOut) { inputLruCache.put(logMessage, timestamp); } else { LOG.debug("Log line filtered out: {} (file: {}, dedupInterval: {}, lastDedupEnabled: {})", logMessage, inputLruCache.getFileName(), inputLruCache.getDedupInterval(), inputLruCache.isLastDedupEnabled()); } } } if (lineMap.containsKey(LogFeederConstants.IN_MEMORY_TIMESTAMP)) { lineMap.remove(LogFeederConstants.IN_MEMORY_TIMESTAMP); } return isLogFilteredOut; }
OutputLineFilter { public Boolean apply(Map<String, Object> lineMap, Input input) { boolean isLogFilteredOut = false; LRUCache inputLruCache = input.getCache(); if (inputLruCache != null && "service".equals(input.getConfigs().get(LogFeederConstants.ROW_TYPE))) { String logMessage = (String) lineMap.get(input.getCacheKeyField()); Long timestamp = null; if (lineMap.containsKey((LogFeederConstants.IN_MEMORY_TIMESTAMP))) { timestamp = (Long) lineMap.get(LogFeederConstants.IN_MEMORY_TIMESTAMP); } if (logMessage != null && timestamp != null) { isLogFilteredOut = !inputLruCache.isEntryReplaceable(logMessage, timestamp); if (!isLogFilteredOut) { inputLruCache.put(logMessage, timestamp); } else { LOG.debug("Log line filtered out: {} (file: {}, dedupInterval: {}, lastDedupEnabled: {})", logMessage, inputLruCache.getFileName(), inputLruCache.getDedupInterval(), inputLruCache.isLastDedupEnabled()); } } } if (lineMap.containsKey(LogFeederConstants.IN_MEMORY_TIMESTAMP)) { lineMap.remove(LogFeederConstants.IN_MEMORY_TIMESTAMP); } return isLogFilteredOut; } }
OutputLineFilter { public Boolean apply(Map<String, Object> lineMap, Input input) { boolean isLogFilteredOut = false; LRUCache inputLruCache = input.getCache(); if (inputLruCache != null && "service".equals(input.getConfigs().get(LogFeederConstants.ROW_TYPE))) { String logMessage = (String) lineMap.get(input.getCacheKeyField()); Long timestamp = null; if (lineMap.containsKey((LogFeederConstants.IN_MEMORY_TIMESTAMP))) { timestamp = (Long) lineMap.get(LogFeederConstants.IN_MEMORY_TIMESTAMP); } if (logMessage != null && timestamp != null) { isLogFilteredOut = !inputLruCache.isEntryReplaceable(logMessage, timestamp); if (!isLogFilteredOut) { inputLruCache.put(logMessage, timestamp); } else { LOG.debug("Log line filtered out: {} (file: {}, dedupInterval: {}, lastDedupEnabled: {})", logMessage, inputLruCache.getFileName(), inputLruCache.getDedupInterval(), inputLruCache.isLastDedupEnabled()); } } } if (lineMap.containsKey(LogFeederConstants.IN_MEMORY_TIMESTAMP)) { lineMap.remove(LogFeederConstants.IN_MEMORY_TIMESTAMP); } return isLogFilteredOut; }  }
OutputLineFilter { public Boolean apply(Map<String, Object> lineMap, Input input) { boolean isLogFilteredOut = false; LRUCache inputLruCache = input.getCache(); if (inputLruCache != null && "service".equals(input.getConfigs().get(LogFeederConstants.ROW_TYPE))) { String logMessage = (String) lineMap.get(input.getCacheKeyField()); Long timestamp = null; if (lineMap.containsKey((LogFeederConstants.IN_MEMORY_TIMESTAMP))) { timestamp = (Long) lineMap.get(LogFeederConstants.IN_MEMORY_TIMESTAMP); } if (logMessage != null && timestamp != null) { isLogFilteredOut = !inputLruCache.isEntryReplaceable(logMessage, timestamp); if (!isLogFilteredOut) { inputLruCache.put(logMessage, timestamp); } else { LOG.debug("Log line filtered out: {} (file: {}, dedupInterval: {}, lastDedupEnabled: {})", logMessage, inputLruCache.getFileName(), inputLruCache.getDedupInterval(), inputLruCache.isLastDedupEnabled()); } } } if (lineMap.containsKey(LogFeederConstants.IN_MEMORY_TIMESTAMP)) { lineMap.remove(LogFeederConstants.IN_MEMORY_TIMESTAMP); } return isLogFilteredOut; }  Boolean apply(Map<String, Object> lineMap, Input input); }
OutputLineFilter { public Boolean apply(Map<String, Object> lineMap, Input input) { boolean isLogFilteredOut = false; LRUCache inputLruCache = input.getCache(); if (inputLruCache != null && "service".equals(input.getConfigs().get(LogFeederConstants.ROW_TYPE))) { String logMessage = (String) lineMap.get(input.getCacheKeyField()); Long timestamp = null; if (lineMap.containsKey((LogFeederConstants.IN_MEMORY_TIMESTAMP))) { timestamp = (Long) lineMap.get(LogFeederConstants.IN_MEMORY_TIMESTAMP); } if (logMessage != null && timestamp != null) { isLogFilteredOut = !inputLruCache.isEntryReplaceable(logMessage, timestamp); if (!isLogFilteredOut) { inputLruCache.put(logMessage, timestamp); } else { LOG.debug("Log line filtered out: {} (file: {}, dedupInterval: {}, lastDedupEnabled: {})", logMessage, inputLruCache.getFileName(), inputLruCache.getDedupInterval(), inputLruCache.isLastDedupEnabled()); } } } if (lineMap.containsKey(LogFeederConstants.IN_MEMORY_TIMESTAMP)) { lineMap.remove(LogFeederConstants.IN_MEMORY_TIMESTAMP); } return isLogFilteredOut; }  Boolean apply(Map<String, Object> lineMap, Input input);  }