@Test public void test_FilterDocuments_SomeAreInvalid() throws SolrServerException, IOException { assertArrayEquals(validRecords.toArray(), solrWriter.filterDocumentIds(harvesterRecords, testBatchId).toArray()); }
public List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId) throws IOException, SolrServerException { final Timer.Context context = PublisherMetrics.Publisher.Read.Solr.solrCheckIdsDurationDuration.time(connectionId); try { if (null == documents || documents.isEmpty()) { return Collections.EMPTY_LIST; } final Set<String> acceptedRecordIds = new HashSet<>(); final List<String> documentIds = new ArrayList<>(); for (final HarvesterRecord document : documents) { documentIds.add(document.getRecordId()); } LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Checking records: " + documents.size()); for (int documentIdsStartChunkIndex = 0; documentIdsStartChunkIndex <= documentIds.size(); documentIdsStartChunkIndex += MAX_NUMBER_OF_IDS_IN_SOLR_QUERY) { final int endOfArray = Math.min(documentIds.size(), documentIdsStartChunkIndex + MAX_NUMBER_OF_IDS_IN_SOLR_QUERY); final List<String> documentIdsToQuery = documentIds.subList(documentIdsStartChunkIndex, endOfArray); if (!documentIdsToQuery.isEmpty()) { final SolrQuery query = new SolrQuery(); final String queryString = "(" + StringUtils.join(documentIdsToQuery, " OR ").replace("/", "\\/") + ")"; query.setRows(MAX_NUMBER_OF_IDS_IN_SOLR_QUERY + 1); query.setQuery("europeana_id:" + queryString); try { final SolrClient server = createServer(publishingBatchId); final QueryResponse response = server.query(query, SolrRequest.METHOD.POST); if (response != null) { final SolrDocumentList solrResults = response.getResults(); for (int resultEntryIndex = 0; resultEntryIndex < solrResults.size(); ++resultEntryIndex) acceptedRecordIds.add(solrResults.get(resultEntryIndex).getFieldValue("europeana_id") .toString()); } server.close(); } catch (Exception e) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Failed when executing existence query for {} documents. Will mark the documents as non-existent. The query is {}", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, queryString); LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "The exception is", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, e ); throw e; } } } final List<HarvesterRecord> filteredDocuments = new ArrayList<>(); for (final HarvesterRecord document: documents) { if (acceptedRecordIds.contains(document.getRecordId())) { filteredDocuments.add(document); } } PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInSolr.inc(filteredDocuments.size()); PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInOneSolr.inc(connectionId, filteredDocuments.size()); return filteredDocuments; } finally { context.close(); } }
SOLRWriter { public List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId) throws IOException, SolrServerException { final Timer.Context context = PublisherMetrics.Publisher.Read.Solr.solrCheckIdsDurationDuration.time(connectionId); try { if (null == documents || documents.isEmpty()) { return Collections.EMPTY_LIST; } final Set<String> acceptedRecordIds = new HashSet<>(); final List<String> documentIds = new ArrayList<>(); for (final HarvesterRecord document : documents) { documentIds.add(document.getRecordId()); } LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Checking records: " + documents.size()); for (int documentIdsStartChunkIndex = 0; documentIdsStartChunkIndex <= documentIds.size(); documentIdsStartChunkIndex += MAX_NUMBER_OF_IDS_IN_SOLR_QUERY) { final int endOfArray = Math.min(documentIds.size(), documentIdsStartChunkIndex + MAX_NUMBER_OF_IDS_IN_SOLR_QUERY); final List<String> documentIdsToQuery = documentIds.subList(documentIdsStartChunkIndex, endOfArray); if (!documentIdsToQuery.isEmpty()) { final SolrQuery query = new SolrQuery(); final String queryString = "(" + StringUtils.join(documentIdsToQuery, " OR ").replace("/", "\\/") + ")"; query.setRows(MAX_NUMBER_OF_IDS_IN_SOLR_QUERY + 1); query.setQuery("europeana_id:" + queryString); try { final SolrClient server = createServer(publishingBatchId); final QueryResponse response = server.query(query, SolrRequest.METHOD.POST); if (response != null) { final SolrDocumentList solrResults = response.getResults(); for (int resultEntryIndex = 0; resultEntryIndex < solrResults.size(); ++resultEntryIndex) acceptedRecordIds.add(solrResults.get(resultEntryIndex).getFieldValue("europeana_id") .toString()); } server.close(); } catch (Exception e) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Failed when executing existence query for {} documents. Will mark the documents as non-existent. The query is {}", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, queryString); LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "The exception is", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, e ); throw e; } } } final List<HarvesterRecord> filteredDocuments = new ArrayList<>(); for (final HarvesterRecord document: documents) { if (acceptedRecordIds.contains(document.getRecordId())) { filteredDocuments.add(document); } } PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInSolr.inc(filteredDocuments.size()); PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInOneSolr.inc(connectionId, filteredDocuments.size()); return filteredDocuments; } finally { context.close(); } } }
SOLRWriter { public List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId) throws IOException, SolrServerException { final Timer.Context context = PublisherMetrics.Publisher.Read.Solr.solrCheckIdsDurationDuration.time(connectionId); try { if (null == documents || documents.isEmpty()) { return Collections.EMPTY_LIST; } final Set<String> acceptedRecordIds = new HashSet<>(); final List<String> documentIds = new ArrayList<>(); for (final HarvesterRecord document : documents) { documentIds.add(document.getRecordId()); } LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Checking records: " + documents.size()); for (int documentIdsStartChunkIndex = 0; documentIdsStartChunkIndex <= documentIds.size(); documentIdsStartChunkIndex += MAX_NUMBER_OF_IDS_IN_SOLR_QUERY) { final int endOfArray = Math.min(documentIds.size(), documentIdsStartChunkIndex + MAX_NUMBER_OF_IDS_IN_SOLR_QUERY); final List<String> documentIdsToQuery = documentIds.subList(documentIdsStartChunkIndex, endOfArray); if (!documentIdsToQuery.isEmpty()) { final SolrQuery query = new SolrQuery(); final String queryString = "(" + StringUtils.join(documentIdsToQuery, " OR ").replace("/", "\\/") + ")"; query.setRows(MAX_NUMBER_OF_IDS_IN_SOLR_QUERY + 1); query.setQuery("europeana_id:" + queryString); try { final SolrClient server = createServer(publishingBatchId); final QueryResponse response = server.query(query, SolrRequest.METHOD.POST); if (response != null) { final SolrDocumentList solrResults = response.getResults(); for (int resultEntryIndex = 0; resultEntryIndex < solrResults.size(); ++resultEntryIndex) acceptedRecordIds.add(solrResults.get(resultEntryIndex).getFieldValue("europeana_id") .toString()); } server.close(); } catch (Exception e) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Failed when executing existence query for {} documents. Will mark the documents as non-existent. The query is {}", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, queryString); LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "The exception is", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, e ); throw e; } } } final List<HarvesterRecord> filteredDocuments = new ArrayList<>(); for (final HarvesterRecord document: documents) { if (acceptedRecordIds.contains(document.getRecordId())) { filteredDocuments.add(document); } } PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInSolr.inc(filteredDocuments.size()); PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInOneSolr.inc(connectionId, filteredDocuments.size()); return filteredDocuments; } finally { context.close(); } } SOLRWriter(final DBTargetConfig solrConfig); }
SOLRWriter { public List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId) throws IOException, SolrServerException { final Timer.Context context = PublisherMetrics.Publisher.Read.Solr.solrCheckIdsDurationDuration.time(connectionId); try { if (null == documents || documents.isEmpty()) { return Collections.EMPTY_LIST; } final Set<String> acceptedRecordIds = new HashSet<>(); final List<String> documentIds = new ArrayList<>(); for (final HarvesterRecord document : documents) { documentIds.add(document.getRecordId()); } LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Checking records: " + documents.size()); for (int documentIdsStartChunkIndex = 0; documentIdsStartChunkIndex <= documentIds.size(); documentIdsStartChunkIndex += MAX_NUMBER_OF_IDS_IN_SOLR_QUERY) { final int endOfArray = Math.min(documentIds.size(), documentIdsStartChunkIndex + MAX_NUMBER_OF_IDS_IN_SOLR_QUERY); final List<String> documentIdsToQuery = documentIds.subList(documentIdsStartChunkIndex, endOfArray); if (!documentIdsToQuery.isEmpty()) { final SolrQuery query = new SolrQuery(); final String queryString = "(" + StringUtils.join(documentIdsToQuery, " OR ").replace("/", "\\/") + ")"; query.setRows(MAX_NUMBER_OF_IDS_IN_SOLR_QUERY + 1); query.setQuery("europeana_id:" + queryString); try { final SolrClient server = createServer(publishingBatchId); final QueryResponse response = server.query(query, SolrRequest.METHOD.POST); if (response != null) { final SolrDocumentList solrResults = response.getResults(); for (int resultEntryIndex = 0; resultEntryIndex < solrResults.size(); ++resultEntryIndex) acceptedRecordIds.add(solrResults.get(resultEntryIndex).getFieldValue("europeana_id") .toString()); } server.close(); } catch (Exception e) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Failed when executing existence query for {} documents. Will mark the documents as non-existent. The query is {}", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, queryString); LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "The exception is", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, e ); throw e; } } } final List<HarvesterRecord> filteredDocuments = new ArrayList<>(); for (final HarvesterRecord document: documents) { if (acceptedRecordIds.contains(document.getRecordId())) { filteredDocuments.add(document); } } PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInSolr.inc(filteredDocuments.size()); PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInOneSolr.inc(connectionId, filteredDocuments.size()); return filteredDocuments; } finally { context.close(); } } SOLRWriter(final DBTargetConfig solrConfig); String getSolrUrl(); void updateDocuments(List<CRFSolrDocument> newDocs, final String publishingBatchId); List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId); String getConnectionId(); }
SOLRWriter { public List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId) throws IOException, SolrServerException { final Timer.Context context = PublisherMetrics.Publisher.Read.Solr.solrCheckIdsDurationDuration.time(connectionId); try { if (null == documents || documents.isEmpty()) { return Collections.EMPTY_LIST; } final Set<String> acceptedRecordIds = new HashSet<>(); final List<String> documentIds = new ArrayList<>(); for (final HarvesterRecord document : documents) { documentIds.add(document.getRecordId()); } LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Checking records: " + documents.size()); for (int documentIdsStartChunkIndex = 0; documentIdsStartChunkIndex <= documentIds.size(); documentIdsStartChunkIndex += MAX_NUMBER_OF_IDS_IN_SOLR_QUERY) { final int endOfArray = Math.min(documentIds.size(), documentIdsStartChunkIndex + MAX_NUMBER_OF_IDS_IN_SOLR_QUERY); final List<String> documentIdsToQuery = documentIds.subList(documentIdsStartChunkIndex, endOfArray); if (!documentIdsToQuery.isEmpty()) { final SolrQuery query = new SolrQuery(); final String queryString = "(" + StringUtils.join(documentIdsToQuery, " OR ").replace("/", "\\/") + ")"; query.setRows(MAX_NUMBER_OF_IDS_IN_SOLR_QUERY + 1); query.setQuery("europeana_id:" + queryString); try { final SolrClient server = createServer(publishingBatchId); final QueryResponse response = server.query(query, SolrRequest.METHOD.POST); if (response != null) { final SolrDocumentList solrResults = response.getResults(); for (int resultEntryIndex = 0; resultEntryIndex < solrResults.size(); ++resultEntryIndex) acceptedRecordIds.add(solrResults.get(resultEntryIndex).getFieldValue("europeana_id") .toString()); } server.close(); } catch (Exception e) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "Failed when executing existence query for {} documents. Will mark the documents as non-existent. The query is {}", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, queryString); LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Migrator.PERSISTENCE_SOLR, publishingBatchId, null, null), "The exception is", MAX_NUMBER_OF_IDS_IN_SOLR_QUERY, e ); throw e; } } } final List<HarvesterRecord> filteredDocuments = new ArrayList<>(); for (final HarvesterRecord document: documents) { if (acceptedRecordIds.contains(document.getRecordId())) { filteredDocuments.add(document); } } PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInSolr.inc(filteredDocuments.size()); PublisherMetrics.Publisher.Read.Solr.totalNumberOfDocumentsThatExistInOneSolr.inc(connectionId, filteredDocuments.size()); return filteredDocuments; } finally { context.close(); } } SOLRWriter(final DBTargetConfig solrConfig); String getSolrUrl(); void updateDocuments(List<CRFSolrDocument> newDocs, final String publishingBatchId); List<HarvesterRecord> filterDocumentIds(final List<HarvesterRecord> documents, final String publishingBatchId); String getConnectionId();  }