class AmbariInfraWithStormLogSearch implements TopologyLogSearch {

    @Override
    public LogSearchResult search(LogSearchCriteria logSearchCriteria) {
        SolrQuery query = new SolrQuery();

        query.setQuery(buildColumnAndValue(COLUMN_NAME_LOG_MESSAGE, buildValue(logSearchCriteria.getSearchString())));
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_TYPE, COLUMN_VALUE_TYPE_WORKER_LOG));
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID, buildValue(logSearchCriteria.getAppId())));
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_TIME, buildDateRangeValue(logSearchCriteria.getFrom(), logSearchCriteria.getTo())));

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

        List<String> logLevels = logSearchCriteria.getLogLevels();
        if (logLevels == null || logLevels.isEmpty()) {
            logLevels = DEFAULT_LOG_LEVELS;
        }
        query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_LEVEL, buildORValues(logLevels)));

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

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

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

        long numFound;
        List<LogSearchResult.LogDocument> 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 logLevel = (String) document.getFieldValue(COLUMN_NAME_LOG_LEVEL);
                String logMessage = (String) document.getFieldValue(COLUMN_NAME_LOG_MESSAGE);
                String host = (String) document.getFieldValue(COLUMN_NAME_HOST);
                String port = (String) document.getFieldValue(COLUMN_NAME_STORM_WORKER_PORT);
                Date logDate = (Date) document.getFieldValue(COLUMN_NAME_LOG_TIME);
                long timestamp = logDate.toInstant().toEpochMilli();

                LogSearchResult.LogDocument logDocument = new LogSearchResult.LogDocument(appId, componentName,
                        logLevel, logMessage, host, port != null ? Integer.parseInt(port) : null, timestamp);
                results.add(logDocument);
            }

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

        return new LogSearchResult(numFound, results);
    }

    public  AmbariInfraWithStormLogSearch();

}