private Set<TimelineEntity> getEntities(Path dir, String entityType,
      TimelineEntityFilters filters, TimelineDataToRetrieve dataToRetrieve)
      throws IOException {
    // First sort the selected entities based on created/start time.
    Map<Long, Set<TimelineEntity>> sortedEntities =
        new TreeMap<>(
            new Comparator<Long>() {
            @Override
            public int compare(Long l1, Long l2) {
              return l2.compareTo(l1);
            }
          }
        );
    dir = getNormalPath(dir);
    if (dir != null) {
      RemoteIterator<LocatedFileStatus> fileStatuses = fs.listFiles(dir,
              false);
      if (fileStatuses != null) {
        while (fileStatuses.hasNext()) {
          LocatedFileStatus locatedFileStatus = fileStatuses.next();
          Path entityFile = locatedFileStatus.getPath();
          if (!entityFile.getName()
              .contains(TIMELINE_SERVICE_STORAGE_EXTENSION)) {
            continue;
          }
          try (BufferedReader reader = new BufferedReader(
              new InputStreamReader(fs.open(entityFile),
                  StandardCharsets.UTF_8))) {
            TimelineEntity entity = readEntityFromFile(reader);
            if (!entity.getType().equals(entityType)) {
              continue;
            }
            if (!isTimeInRange(entity.getCreatedTime(),
                filters.getCreatedTimeBegin(),
                filters.getCreatedTimeEnd())) {
              continue;
            }
            if (filters.getRelatesTo() != null &&
                !filters.getRelatesTo().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchRelatesTo(entity,
                    filters.getRelatesTo())) {
              continue;
            }
            if (filters.getIsRelatedTo() != null &&
                !filters.getIsRelatedTo().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchIsRelatedTo(entity,
                    filters.getIsRelatedTo())) {
              continue;
            }
            if (filters.getInfoFilters() != null &&
                !filters.getInfoFilters().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchInfoFilters(entity,
                    filters.getInfoFilters())) {
              continue;
            }
            if (filters.getConfigFilters() != null &&
                !filters.getConfigFilters().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchConfigFilters(entity,
                    filters.getConfigFilters())) {
              continue;
            }
            if (filters.getMetricFilters() != null &&
                !filters.getMetricFilters().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchMetricFilters(entity,
                    filters.getMetricFilters())) {
              continue;
            }
            if (filters.getEventFilters() != null &&
                !filters.getEventFilters().getFilterList().isEmpty() &&
                !TimelineStorageUtils.matchEventFilters(entity,
                    filters.getEventFilters())) {
              continue;
            }
            TimelineEntity entityToBeReturned = createEntityToBeReturned(
                entity, dataToRetrieve.getFieldsToRetrieve());
            Set<TimelineEntity> entitiesCreatedAtSameTime =
                sortedEntities.get(entityToBeReturned.getCreatedTime());
            if (entitiesCreatedAtSameTime == null) {
              entitiesCreatedAtSameTime = new HashSet<TimelineEntity>();
            }
            entitiesCreatedAtSameTime.add(entityToBeReturned);
            sortedEntities.put(entityToBeReturned.getCreatedTime(),
                entitiesCreatedAtSameTime);
          }
        }
      }
    }

    Set<TimelineEntity> entities = new HashSet<TimelineEntity>();
    long entitiesAdded = 0;
    for (Set<TimelineEntity> entitySet : sortedEntities.values()) {
      for (TimelineEntity entity : entitySet) {
        entities.add(entity);
        ++entitiesAdded;
        if (entitiesAdded >= filters.getLimit()) {
          return entities;
        }
      }
    }
    return entities;
  }
@Test
  void testGetFilteredEntities() throws Exception {
    // Get entities based on info filters.
    TimelineFilterList infoFilterList = new TimelineFilterList();
    infoFilterList.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5));
    Set<TimelineEntity> result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().infoFilters(infoFilterList).build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    // Only one entity with ID id_3 should be returned.
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_3")) {
        fail("Incorrect filtering based on info filters");
      }
    }

    // Get entities based on config filters.
    TimelineFilterList confFilterList = new TimelineFilterList();
    confFilterList.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_1", "123"));
    confFilterList.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "abc"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_3")) {
        fail("Incorrect filtering based on config filters");
      }
    }

    // Get entities based on event filters.
    TimelineFilterList eventFilters = new TimelineFilterList();
    eventFilters.addFilter(
        new TimelineExistsFilter(TimelineCompareOp.EQUAL, "event_2"));
    eventFilters.addFilter(
        new TimelineExistsFilter(TimelineCompareOp.EQUAL, "event_4"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().eventFilters(eventFilters).build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_3")) {
        fail("Incorrect filtering based on event filters");
      }
    }

    // Get entities based on metric filters.
    TimelineFilterList metricFilterList = new TimelineFilterList();
    metricFilterList.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.GREATER_OR_EQUAL, "metric3", 0L));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    // Two entities with IDs' id_1 and id_2 should be returned.
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on metric filters");
      }
    }

    // Get entities based on complex config filters.
    TimelineFilterList list1 = new TimelineFilterList();
    list1.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_1", "129"));
    list1.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "def"));
    TimelineFilterList list2 = new TimelineFilterList();
    list2.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_2", "23"));
    list2.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "abc"));
    TimelineFilterList confFilterList1 =
        new TimelineFilterList(Operator.OR, list1, list2);
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList1)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on config filters");
      }
    }

    TimelineFilterList list3 = new TimelineFilterList();
    list3.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.NOT_EQUAL, "config_1", "123"));
    list3.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.NOT_EQUAL, "config_3", "abc"));
    TimelineFilterList list4 = new TimelineFilterList();
    list4.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_2", "23"));
    TimelineFilterList confFilterList2 =
        new TimelineFilterList(Operator.OR, list3, list4);
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList2)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on config filters");
      }
    }

    TimelineFilterList confFilterList3 = new TimelineFilterList();
    confFilterList3.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.NOT_EQUAL, "config_1", "127"));
    confFilterList3.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.NOT_EQUAL, "config_3", "abc"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList3)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on config filters");
      }
    }

    TimelineFilterList confFilterList4 = new TimelineFilterList();
    confFilterList4.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.EQUAL, "config_dummy", "dummy"));
    confFilterList4.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.EQUAL, "config_3", "def"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList4)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(0, result.size());

    TimelineFilterList confFilterList5 = new TimelineFilterList(Operator.OR);
    confFilterList5.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.EQUAL, "config_dummy", "dummy"));
    confFilterList5.addFilter(new TimelineKeyValueFilter(
        TimelineCompareOp.EQUAL, "config_3", "def"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().configFilters(confFilterList5)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on config filters");
      }
    }

    // Get entities based on complex metric filters.
    TimelineFilterList list6 = new TimelineFilterList();
    list6.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.GREATER_THAN, "metric1", 200));
    list6.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.EQUAL, "metric3", 23));
    TimelineFilterList list7 = new TimelineFilterList();
    list7.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.GREATER_OR_EQUAL, "metric2", 74));
    TimelineFilterList metricFilterList1 =
        new TimelineFilterList(Operator.OR, list6, list7);
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList1)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    // Two entities with IDs' id_2 and id_3 should be returned.
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_2") && !entity.getId().equals("id_3")) {
        fail("Incorrect filtering based on metric filters");
      }
    }

    TimelineFilterList metricFilterList2 = new TimelineFilterList();
    metricFilterList2.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_THAN, "metric2", 70));
    metricFilterList2.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList2)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1")) {
        fail("Incorrect filtering based on metric filters");
      }
    }

    TimelineFilterList metricFilterList3 = new TimelineFilterList();
    metricFilterList3.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_THAN, "dummy_metric", 30));
    metricFilterList3.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList3)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(0, result.size());

    TimelineFilterList metricFilterList4 = new TimelineFilterList(Operator.OR);
    metricFilterList4.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_THAN, "dummy_metric", 30));
    metricFilterList4.addFilter(new TimelineCompareFilter(
        TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList4)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on metric filters");
      }
    }

    TimelineFilterList metricFilterList5 =
        new TimelineFilterList(new TimelineCompareFilter(
            TimelineCompareOp.NOT_EQUAL, "metric2", 74));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().metricFilters(metricFilterList5)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) {
        fail("Incorrect filtering based on metric filters");
      }
    }

    TimelineFilterList infoFilterList1 = new TimelineFilterList();
    infoFilterList1.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5));
    infoFilterList1.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.NOT_EQUAL, "info4", 20));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().infoFilters(infoFilterList1)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(0, result.size());

    TimelineFilterList infoFilterList2 = new TimelineFilterList(Operator.OR);
    infoFilterList2.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5));
    infoFilterList2.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info1", "val1"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().infoFilters(infoFilterList2)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(2, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1") && !entity.getId().equals("id_3")) {
        fail("Incorrect filtering based on info filters");
      }
    }

    TimelineFilterList infoFilterList3 = new TimelineFilterList();
    infoFilterList3.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "dummy_info", 1));
    infoFilterList3.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", "val5"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().infoFilters(infoFilterList3)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(0, result.size());

    TimelineFilterList infoFilterList4 = new TimelineFilterList(Operator.OR);
    infoFilterList4.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "dummy_info", 1));
    infoFilterList4.addFilter(
        new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", "val5"));
    result = reader.getEntities(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", null),
        new TimelineEntityFilters.Builder().infoFilters(infoFilterList4)
            .build(),
        new TimelineDataToRetrieve());
    assertEquals(1, result.size());
    for (TimelineEntity entity : result) {
      if (!entity.getId().equals("id_1")) {
        fail("Incorrect filtering based on info filters");
      }
    }
  }