class AmbariInfraWithStormLogSearch implements TopologyLogSearch {

    @Override
    public EventSearchResult searchEvent(EventSearchCriteria criteria) {
        SolrQuery query = new SolrQuery();

        String searchString = criteria.getSearchString();

        List<String> queryStrings = new ArrayList<>();
        addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES);
        addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_HEADERS);
        addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES);

        // this is to get rid of non-streamline events
        String queryString = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ID, buildValue(null));

        if (!queryStrings.isEmpty()) {
            queryString += " AND (" + String.join(" OR ", queryStrings) + ")";
        }

        query.setQuery(queryString);

        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_TYPE, COLUMN_VALUE_TYPE_EVENT));
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID, buildValue(criteria.getAppId())));
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_TIME, buildDateRangeValue(criteria.getFrom(), criteria.getTo())));

        List<String> componentNames = criteria.getComponentNames();
        if (componentNames != null && !componentNames.isEmpty()) {
            query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME, buildORValues(componentNames)));
        }

        String searchEventId = criteria.getSearchEventId();
        if (searchEventId != null) {
            // eventId OR rootId OR parentId
            String queryToEventId = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ID, buildValue(searchEventId));
            String queryToRootIds = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET, buildValue("*" + searchEventId + "*"));
            String queryToParentIds = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET, buildValue("*" + searchEventId + "*"));
            query.addFilterQuery(queryToEventId + " OR " + queryToRootIds + " OR " + queryToParentIds);
        }

        if (criteria.getAscending() == null || criteria.getAscending()) {
            query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.asc);
        } else {
            query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.desc);
        }

        if (criteria.getStart() != null) {
            query.setStart(criteria.getStart());
        }
        if (criteria.getLimit() != null) {
            query.setRows(criteria.getLimit());
        }

        LOG.debug("Querying to Solr: query => {}", query);

        long numFound;
        List<EventSearchResult.Event> results = new ArrayList<>();
        try {
            QueryResponse response = solr.query(query);

            SolrDocumentList docList = response.getResults();
            numFound = docList.getNumFound();

            for (SolrDocument document : docList) {
                String appId = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID);
                String componentName = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME);
                String eventId = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_ID);
                String rootIds = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET);
                String parentIds = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET);
                String keyValues = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES);
                String headers = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_HEADERS);
                String auxKeyValues = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES);

                Date logDate = (Date) document.getFieldValue(COLUMN_NAME_LOG_TIME);
                long timestamp = logDate.toInstant().toEpochMilli();

                EventSearchResult.Event event = new EventSearchResult.Event(appId, componentName,
                        eventId, rootIds, parentIds, keyValues, headers, auxKeyValues, timestamp);
                results.add(event);
            }

        } catch (SolrServerException | IOException e) {
            // TODO: any fine-grained control needed?
            throw new RuntimeException(e);
        }

        return new EventSearchResult(numFound, results);
    }

    public  AmbariInfraWithStormLogSearch();

}