class OutputLineFilter {

  public Boolean apply(Map<String, Object> lineMap, Input input) {
    boolean isLogFilteredOut = false;
    LRUCache inputLruCache = input.getCache();
    if (inputLruCache != null && "service".equals(input.getInputDescriptor().getRowtype())) {
      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;
  }

}