virtualFields)
- {
- this.virtualFields = virtualFields;
- }
-
-
-
- /*******************************************************************************
- * Fluent setter for virtualFields
- *
- * @param virtualFields
- * Map of virtual fields - keyed by the virtual field's name, value is meta data
- * for the virtual field.
- * Virtual fields do not exist in the backend system. Rather, they
- * are (or at least tend to be) computed from other fields - e.g., math operations
- * (like a percentage based on 2 numbers), or concatenation of strings, or mappings
- * from e.g., booleans to strings - or more complex logic based on multiple fields
- * @return this
- *******************************************************************************/
- public QTableMetaData withVirtualFields(Map virtualFields)
- {
- this.virtualFields = virtualFields;
- return (this);
- }
-
-}
diff --git a/pom.xml b/pom.xml
index 1a09b00..d090657 100644
--- a/pom.xml
+++ b/pom.xml
@@ -27,7 +27,7 @@
com.kingsrook
qbit-build-parent
- 1.6.0
+ 2.0.0
com.kingsrook.qbits
@@ -75,13 +75,6 @@
pom
import
-
- com.kingsrook.qqq
- qqq-bom-pom
- 0.40.0
- pom
- import
-
@@ -97,16 +90,16 @@
${opensearch.version}
+
+
org.apache.httpcomponents.client5
httpclient5
- 5.3
com.fasterxml.jackson.datatype
jackson-datatype-jsr310
- 2.21.0
@@ -147,13 +140,13 @@
org.testcontainers
testcontainers
- 1.21.0
+ 1.21.4
test
org.testcontainers
junit-jupiter
- 1.21.0
+ 1.21.4
test
@@ -224,4 +217,39 @@
+
+
+
+
+
+
+ qqq-snapshot
+
+ 4.1.0-SNAPSHOT
+
+
+
+
+ com.kingsrook.qqq
+ qqq-bom-pom
+ ${qqq.snapshot.version}
+ pom
+ import
+
+
+
+
+
+ sonatype-snapshots
+ https://central.sonatype.com/repository/maven-snapshots/
+
+ false
+
+
+ true
+
+
+
+
+
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducer.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducer.java
index 794c21b..3a6a28e 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducer.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducer.java
@@ -43,6 +43,7 @@ import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import com.kingsrook.qbits.quicksearch.processes.BasepullIndexStep;
import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
+import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher;
@@ -62,6 +63,7 @@ public class QuickSearchQBitProducer
public static final String BASEPULL_PROCESS_NAME = "quickSearchBasepullIndex";
public static final String FULL_REINDEX_PROCESS_NAME = "quickSearchFullReindex";
+ public static final String RECONCILE_PROCESS_NAME = "quickSearchReconcileIndex";
public static final String APP_NAME = "quickSearchAdmin";
private QuickSearchQBitConfig config;
@@ -217,14 +219,17 @@ public class QuickSearchQBitProducer
////////////////////////////////////////////////////
String basepullProcessName = config.applyPrefix(BASEPULL_PROCESS_NAME);
String fullReindexProcessName = config.applyPrefix(FULL_REINDEX_PROCESS_NAME);
+ String reconcileProcessName = config.applyPrefix(RECONCILE_PROCESS_NAME);
QProcessMetaData basepullProcess = buildBasepullProcess(basepullProcessName);
QProcessMetaData fullReindexProcess = buildFullReindexProcess(fullReindexProcessName);
+ QProcessMetaData reconcileProcess = buildReconcileProcess(reconcileProcessName);
if(Boolean.TRUE.equals(config.getEnableScheduledProcesses()))
{
qInstance.addProcess(basepullProcess);
qInstance.addProcess(fullReindexProcess);
+ qInstance.addProcess(reconcileProcess);
}
////////////////////////////////////////////////////
@@ -249,6 +254,7 @@ public class QuickSearchQBitProducer
{
app.withChild(basepullProcess);
app.withChild(fullReindexProcess);
+ app.withChild(reconcileProcess);
}
qInstance.addApp(app);
@@ -527,6 +533,23 @@ public class QuickSearchQBitProducer
+ /***************************************************************************
+ ** Build the reconcile process metadata: re-index each table from its
+ ** source and remove documents with no source record, without wiping the
+ ** index first. Takes the same optional tableName input as full reindex.
+ ***************************************************************************/
+ private QProcessMetaData buildReconcileProcess(String processName)
+ {
+ return new QProcessMetaData()
+ .withName(processName)
+ .withLabel("Quick Search Reconcile Index")
+ .withStep(new QBackendStepMetaData()
+ .withName("reconcile")
+ .withCode(new QCodeReference(ReconcileIndexStep.class)));
+ }
+
+
+
/***************************************************************************
** Register post-insert, post-update, and post-delete customizers on each
** discovered source table.
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizer.java b/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizer.java
index c204f62..3b80ddc 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizer.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizer.java
@@ -24,6 +24,7 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -56,8 +57,8 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
/*******************************************************************************
- ** Post-delete hook: builds DELETE events for each deleted record and publishes
- ** them via the configured IndexEventPublisher.
+ ** Post-delete hook: builds DELETE events for each successfully deleted
+ ** record and publishes them via the configured IndexEventPublisher.
**
** If the publisher is null (QBit not yet initialized) or if publishing fails,
** a warning is logged and the original records are returned without throwing,
@@ -87,13 +88,26 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
List events = new ArrayList<>();
for(QRecord record : records)
{
+ ///////////////////////////////////////////////////////////////////
+ // a record with errors was not deleted, so its document stays; //
+ // one without a primary key has no document id to remove //
+ ///////////////////////////////////////////////////////////////////
+ String recordId = record.getValueString(primaryKeyField);
+ if(recordId == null || CollectionUtils.nullSafeHasContents(record.getErrors()))
+ {
+ continue;
+ }
+
events.add(new IndexEvent()
.withAction(IndexEventAction.DELETE)
.withTableName(tableName)
- .withRecordId(record.getValueString(primaryKeyField)));
+ .withRecordId(recordId));
}
- publisher.publishDeleteEvents(events);
+ if(!events.isEmpty())
+ {
+ publisher.publishDeleteEvents(events);
+ }
}
catch(Exception e)
{
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizer.java b/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizer.java
index 307d764..81b51fc 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizer.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizer.java
@@ -17,14 +17,21 @@
package com.kingsrook.qbits.quicksearch.customizers;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -37,6 +44,11 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
** Post-update table customizer that re-indexes updated records in the Quick
** Search OpenSearch index.
**
+ ** An update's records may hold only the fields that changed, so each updated
+ ** record is re-read from the source table (in the update's transaction) and
+ ** the full record is indexed. Indexing the sparse record would drop the
+ ** unchanged fields from the document.
+ **
** QQQ instantiates this class via reflection using the no-arg constructor.
** The publisher and table configuration are obtained from the static
** QuickSearchQBitContext at call time.
@@ -57,8 +69,10 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
/*******************************************************************************
- ** Post-update hook: builds INDEX events for each updated record and publishes
- ** them via the configured IndexEventPublisher.
+ ** Post-update hook: re-reads the successfully updated records from the
+ ** source table, builds INDEX events from them, and publishes them via the
+ ** configured IndexEventPublisher. Records with errors were not updated, so
+ ** they are skipped.
**
** If the publisher is null (QBit not yet initialized) or if publishing fails,
** a warning is logged and the original records are returned without throwing,
@@ -86,7 +100,7 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
String primaryKeyField = (tableConfig != null && tableConfig.getPrimaryKeyField() != null) ? tableConfig.getPrimaryKeyField() : "id";
List events = new ArrayList<>();
- for(QRecord record : records)
+ for(QRecord record : fetchCurrentRecords(updateInput, records, primaryKeyField))
{
events.add(new IndexEvent()
.withAction(IndexEventAction.INDEX)
@@ -95,7 +109,10 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
.withRecord(record));
}
- publisher.publishIndexEvents(events);
+ if(!events.isEmpty())
+ {
+ publisher.publishIndexEvents(events);
+ }
}
catch(Exception e)
{
@@ -105,4 +122,36 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
return records;
}
+
+
+ /*******************************************************************************
+ ** Query the source table for the full, current version of each successfully
+ ** updated record. Uses the update's transaction, so uncommitted changes are
+ ** visible.
+ *******************************************************************************/
+ private List fetchCurrentRecords(UpdateInput updateInput, List updatedRecords, String primaryKeyField) throws QException
+ {
+ List primaryKeys = new ArrayList<>();
+ for(QRecord record : updatedRecords)
+ {
+ Serializable primaryKey = record.getValue(primaryKeyField);
+ if(primaryKey != null && CollectionUtils.nullSafeIsEmpty(record.getErrors()))
+ {
+ primaryKeys.add(primaryKey);
+ }
+ }
+
+ if(primaryKeys.isEmpty())
+ {
+ return (new ArrayList<>());
+ }
+
+ QueryInput queryInput = new QueryInput();
+ queryInput.setTableName(updateInput.getTableName());
+ queryInput.setTransaction(updateInput.getTransaction());
+ queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, primaryKeys)));
+
+ return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
+ }
+
}
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/OpenSearchDocument.java b/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/OpenSearchDocument.java
index e8acff8..0244685 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/OpenSearchDocument.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/OpenSearchDocument.java
@@ -172,6 +172,16 @@ public class OpenSearchDocument
*******************************************************************************/
@JsonIgnore
public String getDocumentId()
+ {
+ return (buildDocumentId(sourceTable, recordId));
+ }
+
+
+
+ /*******************************************************************************
+ ** Build the composite OpenSearch _id for a source table and record ID.
+ *******************************************************************************/
+ public static String buildDocumentId(String sourceTable, String recordId)
{
return (sourceTable + ":" + recordId);
}
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClient.java b/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClient.java
index 6f6a5e1..32241cb 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClient.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClient.java
@@ -19,6 +19,7 @@ package com.kingsrook.qbits.quicksearch.opensearch;
import java.io.Closeable;
import java.io.IOException;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -32,8 +33,10 @@ import org.apache.hc.client5.http.auth.AuthScope;
import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
import org.apache.hc.core5.http.HttpHost;
+import org.opensearch.client.json.JsonData;
import org.opensearch.client.json.jackson.JacksonJsonpMapper;
import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch._types.Conflicts;
import org.opensearch.client.opensearch._types.FieldValue;
import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
import org.opensearch.client.opensearch._types.query_dsl.MultiMatchQuery;
@@ -42,12 +45,14 @@ import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
import org.opensearch.client.opensearch.core.BulkRequest;
import org.opensearch.client.opensearch.core.BulkResponse;
import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
+import org.opensearch.client.opensearch.core.DeleteByQueryResponse;
import org.opensearch.client.opensearch.core.DeleteRequest;
import org.opensearch.client.opensearch.core.IndexRequest;
import org.opensearch.client.opensearch.core.SearchRequest;
import org.opensearch.client.opensearch.core.SearchResponse;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
+import org.opensearch.client.opensearch.core.bulk.DeleteOperation;
import org.opensearch.client.opensearch.core.bulk.IndexOperation;
import org.opensearch.client.opensearch.core.search.Highlight;
import org.opensearch.client.opensearch.core.search.HighlightField;
@@ -104,17 +109,33 @@ public class QuickSearchOpenSearchClient implements Closeable
String username = config.getOpensearchUsername();
String password = config.getOpensearchPassword();
+ BasicCredentialsProvider credentialsProvider = null;
if(username != null && password != null)
{
- BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(
new AuthScope(httpHost),
new UsernamePasswordCredentials(username, password.toCharArray()));
-
- builder.setHttpClientConfigCallback(httpClientBuilder ->
- httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
}
+ BasicCredentialsProvider finalCredentialsProvider = credentialsProvider;
+ builder.setHttpClientConfigCallback(httpClientBuilder ->
+ {
+ ///////////////////////////////////////////////////////////////////////
+ // httpclient5 5.6+ decompresses gzip responses itself but leaves //
+ // the Content-Encoding header, so the OpenSearch transport would //
+ // gunzip the body a second time and fail. Let the transport alone //
+ // handle compression. //
+ ///////////////////////////////////////////////////////////////////////
+ httpClientBuilder.disableContentCompression();
+
+ if(finalCredentialsProvider != null)
+ {
+ httpClientBuilder.setDefaultCredentialsProvider(finalCredentialsProvider);
+ }
+ return (httpClientBuilder);
+ });
+
transport = builder.build();
client = new OpenSearchClient(transport);
indexName = config.getOpensearchIndexName();
@@ -246,28 +267,7 @@ public class QuickSearchOpenSearchClient implements Closeable
.document(doc)))));
}
- BulkRequest bulkRequest = BulkRequest.of(r -> r.operations(operations));
-
- try
- {
- BulkResponse response = client.bulk(bulkRequest);
-
- for(BulkResponseItem item : response.items())
- {
- if(item.error() != null)
- {
- result.addFailure(item.id() + ": " + item.error().reason());
- }
- else
- {
- result.addSuccess();
- }
- }
- }
- catch(IOException e)
- {
- throw new QException("Bulk index request failed: " + e.getMessage(), e);
- }
+ executeBulk(operations, result, "index");
}
return (result);
@@ -275,6 +275,85 @@ public class QuickSearchOpenSearchClient implements Closeable
+ /*******************************************************************************
+ ** Bulk-delete documents by their composite document IDs, in batches of at
+ ** most batchSize.
+ **
+ ** Every document is attempted; one failing item does not stop the others.
+ ** A document that is not in the index counts as a success, because it is
+ ** already absent. Per-item errors are recorded via addFailure().
+ **
+ ** @param documentIds the composite document IDs; null is treated as empty
+ ** @param batchSize maximum number of documents per bulk request
+ ** @return accumulated result across all batches
+ ** @throws QException if any batch request fails at the transport level
+ *******************************************************************************/
+ public BulkIndexResult deleteDocuments(List documentIds, int batchSize) throws QException
+ {
+ BulkIndexResult result = new BulkIndexResult();
+
+ if(documentIds == null || documentIds.isEmpty())
+ {
+ return (result);
+ }
+
+ int total = documentIds.size();
+
+ for(int start = 0; start < total; start += batchSize)
+ {
+ List batch = documentIds.subList(start, Math.min(start + batchSize, total));
+ List operations = new ArrayList<>();
+
+ for(String documentId : batch)
+ {
+ operations.add(BulkOperation.of(o -> o
+ .delete(DeleteOperation.of(d -> d
+ .index(indexName)
+ .id(documentId)))));
+ }
+
+ executeBulk(operations, result, "delete");
+ }
+
+ return (result);
+ }
+
+
+
+ /*******************************************************************************
+ ** Send one bulk request and record each item's outcome in result.
+ **
+ ** @param operations the bulk operations to send
+ ** @param result accumulator for per-item successes and failures
+ ** @param operationName used in the transport-failure message
+ ** @throws QException if the request fails at the transport level
+ *******************************************************************************/
+ private void executeBulk(List operations, BulkIndexResult result, String operationName) throws QException
+ {
+ try
+ {
+ BulkResponse response = client.bulk(BulkRequest.of(r -> r.operations(operations)));
+
+ for(BulkResponseItem item : response.items())
+ {
+ if(item.error() != null)
+ {
+ result.addFailure(item.id() + ": " + item.error().reason());
+ }
+ else
+ {
+ result.addSuccess();
+ }
+ }
+ }
+ catch(IOException e)
+ {
+ throw new QException("Bulk " + operationName + " request failed: " + e.getMessage(), e);
+ }
+ }
+
+
+
/*******************************************************************************
** Delete a single document identified by sourceTable and recordId.
**
@@ -284,7 +363,7 @@ public class QuickSearchOpenSearchClient implements Closeable
*******************************************************************************/
public void deleteDocument(String sourceTable, String recordId) throws QException
{
- String documentId = sourceTable + ":" + recordId;
+ String documentId = OpenSearchDocument.buildDocumentId(sourceTable, recordId);
try
{
@@ -325,6 +404,48 @@ public class QuickSearchOpenSearchClient implements Closeable
+ /*******************************************************************************
+ ** Delete a table's documents that were not (re-)indexed at or after cutoff,
+ ** including any that have no indexedAt value.
+ **
+ ** Only documents visible to search are considered, so refresh the index
+ ** first. Documents changed while the delete runs are skipped (version
+ ** conflicts proceed), since they were just re-indexed.
+ **
+ ** @param sourceTable the table whose documents to consider
+ ** @param cutoff documents indexed before this instant are deleted
+ ** @return the number of documents deleted
+ ** @throws QException if the OpenSearch call fails
+ *******************************************************************************/
+ public Long deleteDocumentsIndexedBefore(String sourceTable, Instant cutoff) throws QException
+ {
+ try
+ {
+ DeleteByQueryResponse response = client.deleteByQuery(DeleteByQueryRequest.of(r -> r
+ .index(indexName)
+ .conflicts(Conflicts.Proceed)
+ .refresh(true)
+ .query(Query.of(q -> q
+ .bool(b -> b
+ .filter(f -> f
+ .term(t -> t
+ .field("sourceTable")
+ .value(FieldValue.of(sourceTable))))
+ .mustNot(mn -> mn
+ .range(rg -> rg
+ .field("indexedAt")
+ .gte(JsonData.of(cutoff.toString())))))))));
+
+ return (response.deleted() == null ? 0L : response.deleted());
+ }
+ catch(IOException e)
+ {
+ throw new QException("Failed to delete stale documents for table [" + sourceTable + "]: " + e.getMessage(), e);
+ }
+ }
+
+
+
/*******************************************************************************
** Execute a boosted multi-match search across the index.
**
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java
new file mode 100644
index 0000000..61515a6
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java
@@ -0,0 +1,280 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch.processes;
+
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
+import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
+import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
+import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
+import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** Process step that reconciles the search index with its source tables.
+ **
+ ** For each table (or only the one named by the optional "tableName" input):
+ ** 1. Re-index every source record, paging in primary-key order so each record
+ ** is read exactly once, even while other rows are inserted or deleted.
+ ** 2. If every document indexed, refresh the index and delete the table's
+ ** documents that this run did not re-index (indexedAt before the run
+ ** started). Those have no source record: their deletes were missed.
+ **
+ ** Unlike FullReindexStep, the table's documents are never wiped first, so
+ ** search keeps working during the run. When any document fails to index, the
+ ** stale-document delete is skipped (it would remove that record's old
+ ** document), and the run is marked FAILED.
+ **
+ ** The cutoff comes from this JVM's clock, as do real-time indexedAt values on
+ ** this node; application nodes are assumed to keep their clocks in sync.
+ *******************************************************************************/
+public class ReconcileIndexStep extends AbstractIndexingStep
+{
+ private static final QLogger LOG = QLogger.getLogger(ReconcileIndexStep.class);
+
+ public static final String RUN_TYPE = "RECONCILE";
+
+
+
+ /*******************************************************************************
+ ** Totals from reconciling one table.
+ *******************************************************************************/
+ private record ReconcileCounts(Integer recordsIndexed, Long documentsRemoved)
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Reconcile every discovered table, or only the one named by "tableName".
+ **
+ ** Output values: recordsIndexed (Integer) and documentsRemoved (Long),
+ ** summed across tables.
+ **
+ ** @throws QException if reconciling a table fails (its run is marked FAILED)
+ *******************************************************************************/
+ @Override
+ public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+ {
+ String tableNameFilter = input.getValueString("tableName");
+
+ Integer recordsIndexed = 0;
+ Long documentsRemoved = 0L;
+
+ for(QuickSearchableTableConfig tableConfig : getDiscoveredTables())
+ {
+ if(tableNameFilter != null && !tableNameFilter.isBlank() && !tableNameFilter.equals(tableConfig.getTableName()))
+ {
+ continue;
+ }
+
+ ensureIndexRowExists(tableConfig.getTableName(), tableConfig);
+ ReconcileCounts counts = reconcileTable(tableConfig);
+
+ recordsIndexed = recordsIndexed + counts.recordsIndexed();
+ documentsRemoved = documentsRemoved + counts.documentsRemoved();
+ }
+
+ output.addValue("recordsIndexed", recordsIndexed);
+ output.addValue("documentsRemoved", documentsRemoved);
+ }
+
+
+
+ /*******************************************************************************
+ ** Reconcile one table, recording the run in quickSearchIndexRun.
+ *******************************************************************************/
+ private ReconcileCounts reconcileTable(QuickSearchableTableConfig tableConfig) throws QException
+ {
+ String tableName = tableConfig.getTableName();
+ String primaryKeyField = tableConfig.getPrimaryKeyField();
+ QuickSearchOpenSearchClient client = getClient();
+ QuickSearchQBitConfig config = getConfig();
+
+ Integer indexId = queryIndexId(tableName);
+ QuickSearchIndexRun run = createRunRecord(indexId, RUN_TYPE);
+
+ Integer totalProcessed = 0;
+ Integer totalIndexed = 0;
+ Integer totalErrors = 0;
+ String firstError = null;
+
+ try
+ {
+ ///////////////////////////////////////////////////////////////////////
+ // every document this run writes has indexedAt at or after this //
+ // instant, so afterwards any older document has no source record //
+ ///////////////////////////////////////////////////////////////////////
+ Instant reconcileStart = Instant.now();
+
+ Serializable lastPrimaryKey = null;
+ while(true)
+ {
+ List batch = querySourceBatchAfter(tableName, primaryKeyField, lastPrimaryKey, config.getSourceBatchSize());
+ if(batch.isEmpty())
+ {
+ break;
+ }
+
+ List documents = new ArrayList<>();
+ for(QRecord record : batch)
+ {
+ OpenSearchDocument document = IndexingUtils.buildDocument(record, tableConfig);
+ if(document != null)
+ {
+ documents.add(document);
+ }
+ }
+
+ if(!documents.isEmpty())
+ {
+ BulkIndexResult result = client.indexDocuments(documents, config.getBulkBatchSize());
+ totalIndexed = totalIndexed + result.getSuccessCount();
+ totalErrors = totalErrors + result.getFailureCount();
+
+ if(firstError == null && !result.getErrors().isEmpty())
+ {
+ firstError = result.getErrors().get(0);
+ }
+ }
+
+ totalProcessed = totalProcessed + batch.size();
+ lastPrimaryKey = batch.get(batch.size() - 1).getValue(primaryKeyField);
+
+ if(lastPrimaryKey == null)
+ {
+ //////////////////////////////////////////////////////////////
+ // paging needs a key to continue after; without one, the //
+ // next query would start over and never finish //
+ //////////////////////////////////////////////////////////////
+ throw new QException("Source record has no value in primary key field [" + primaryKeyField + "]");
+ }
+ }
+
+ Long documentsRemoved = 0L;
+ if(totalErrors.equals(0))
+ {
+ client.refreshIndex();
+ documentsRemoved = client.deleteDocumentsIndexedBefore(tableName, reconcileStart);
+ updateIndexRow(indexId, reconcileStart, totalProcessed);
+ }
+ else
+ {
+ LOG.warn("Reconcile had indexing errors; stale documents were not removed", logPair("tableName", tableName), logPair("errorCount", totalErrors));
+ }
+
+ String finalStatus = totalErrors.equals(0) ? "COMPLETED" : "FAILED";
+ completeRunRecord(run, finalStatus, totalProcessed, totalIndexed, totalErrors, firstError);
+
+ LOG.info("Reconcile complete", logPair("tableName", tableName), logPair("totalProcessed", totalProcessed),
+ logPair("totalIndexed", totalIndexed), logPair("totalErrors", totalErrors), logPair("documentsRemoved", documentsRemoved));
+
+ return (new ReconcileCounts(totalIndexed, documentsRemoved));
+ }
+ catch(Exception e)
+ {
+ LOG.warn("Reconcile failed", e, logPair("tableName", tableName));
+ completeRunRecord(run, "FAILED", totalProcessed, totalIndexed, totalErrors, e.getMessage());
+ throw new QException("Reconcile failed for table [" + tableName + "]: " + e.getMessage(), e);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Query the next batch of source records, in primary-key order, after the
+ ** given key (or from the start when it is null).
+ *******************************************************************************/
+ private List querySourceBatchAfter(String tableName, String primaryKeyField, Serializable afterPrimaryKey, Integer limit) throws QException
+ {
+ QQueryFilter filter = new QQueryFilter()
+ .withOrderBy(new QFilterOrderBy(primaryKeyField, true))
+ .withLimit(limit);
+
+ if(afterPrimaryKey != null)
+ {
+ filter.withCriteria(new QFilterCriteria(primaryKeyField, QCriteriaOperator.GREATER_THAN, afterPrimaryKey));
+ }
+
+ QueryInput queryInput = new QueryInput();
+ queryInput.setTableName(tableName);
+ queryInput.setFilter(filter);
+
+ return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
+ }
+
+
+
+ /*******************************************************************************
+ ** Look up the ID of the quickSearchIndex row for a table.
+ *******************************************************************************/
+ private Integer queryIndexId(String tableName) throws QException
+ {
+ QueryInput queryInput = new QueryInput();
+ queryInput.setTableName(getConfig().getQuickSearchIndexTableName());
+ queryInput.setFilter(new QQueryFilter()
+ .withCriteria(new QFilterCriteria("tableName", QCriteriaOperator.EQUALS, tableName)));
+
+ List rows = new QueryAction().execute(queryInput).getRecords();
+ if(CollectionUtils.nullSafeIsEmpty(rows))
+ {
+ throw new QException("No QuickSearchIndex row found for table: " + tableName);
+ }
+
+ return (rows.get(0).getValueInteger("id"));
+ }
+
+
+
+ /*******************************************************************************
+ ** Record a successful reconcile on the quickSearchIndex row. Records changed
+ ** after reconcileStart may not have been read, so basepull resumes from there.
+ *******************************************************************************/
+ private void updateIndexRow(Integer indexId, Instant reconcileStart, Integer recordCount) throws QException
+ {
+ UpdateInput updateInput = new UpdateInput();
+ updateInput.setTableName(getConfig().getQuickSearchIndexTableName());
+ updateInput.setRecords(List.of(new QRecord()
+ .withValue("id", indexId)
+ .withValue("lastFullReindexTime", Instant.now())
+ .withValue("lastBasepullTime", reconcileStart)
+ .withValue("recordCount", recordCount)));
+
+ new UpdateAction().execute(updateInput);
+ }
+
+}
diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisher.java b/src/main/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisher.java
index a5dc859..cf4ca85 100644
--- a/src/main/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisher.java
+++ b/src/main/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisher.java
@@ -24,6 +24,7 @@ import java.util.Map;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
+import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import com.kingsrook.qbits.quicksearch.processes.IndexingUtils;
@@ -35,13 +36,15 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
** immediately against the OpenSearch client.
**
** Index events are grouped by table, converted to OpenSearchDocument instances
- ** using IndexingUtils, and bulk-indexed. Delete events are processed
- ** individually.
+ ** using IndexingUtils, and bulk-indexed. Delete events are bulk-deleted.
+ ** Either way, per-document failures are raised as a QException.
*******************************************************************************/
public class SynchronousIndexEventPublisher implements IndexEventPublisher
{
private static final QLogger LOG = QLogger.getLogger(SynchronousIndexEventPublisher.class);
+ private static final Integer MAX_ERRORS_IN_MESSAGE = 10;
+
private final QuickSearchOpenSearchClient client;
private final Map tableConfigsByName;
private final Integer bulkBatchSize;
@@ -119,14 +122,16 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
if(!allDocuments.isEmpty())
{
- client.indexDocuments(allDocuments, bulkBatchSize);
+ BulkIndexResult result = client.indexDocuments(allDocuments, bulkBatchSize);
+ throwIfAnyFailed(result, "index", allDocuments.size());
}
}
/*******************************************************************************
- ** Publish delete events synchronously by calling deleteDocument for each event.
+ ** Publish delete events synchronously as one bulk delete, so every document
+ ** is attempted even when some fail. Throws if any document failed.
*******************************************************************************/
@Override
public void publishDeleteEvents(List events) throws QException
@@ -136,10 +141,33 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
return;
}
+ List documentIds = new ArrayList<>();
for(IndexEvent event : events)
{
- client.deleteDocument(event.getTableName(), event.getRecordId());
+ documentIds.add(OpenSearchDocument.buildDocumentId(event.getTableName(), event.getRecordId()));
}
+
+ BulkIndexResult result = client.deleteDocuments(documentIds, bulkBatchSize);
+ throwIfAnyFailed(result, "delete", documentIds.size());
+ }
+
+
+
+ /*******************************************************************************
+ ** Throw a QException naming the failed documents, if any bulk item failed,
+ ** so failures reach the caller instead of leaving the index silently stale.
+ *******************************************************************************/
+ private void throwIfAnyFailed(BulkIndexResult result, String operationName, Integer attemptedCount) throws QException
+ {
+ if(result == null || Boolean.TRUE.equals(result.isFullySuccessful()))
+ {
+ return;
+ }
+
+ List errors = result.getErrors();
+ List errorsShown = errors.subList(0, Math.min(errors.size(), MAX_ERRORS_IN_MESSAGE));
+ throw (new QException("Failed to " + operationName + " " + result.getFailureCount() + " of " + attemptedCount
+ + " search documents: " + String.join("; ", errorsShown)));
}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/BaseQuickSearchTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/BaseQuickSearchTest.java
index 903d7e7..1a42732 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/BaseQuickSearchTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/BaseQuickSearchTest.java
@@ -79,7 +79,7 @@ public class BaseQuickSearchTest
////////////////////////////////////////////////////
// Set up fully-anonymous authentication //
////////////////////////////////////////////////////
- qInstance.setAuthentication(new QAuthenticationMetaData()
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/ConfigDrivenIndexingIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/ConfigDrivenIndexingIntegrationTest.java
index 0fb8e72..e166398 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/ConfigDrivenIndexingIntegrationTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/ConfigDrivenIndexingIntegrationTest.java
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
@@ -61,7 +62,8 @@ import static org.assertj.core.api.Assertions.assertThat;
** records, runs FullReindexStep, and verifies search results including custom
** recordLabelFormat.
*******************************************************************************/
-@Testcontainers(disabledWithoutDocker = true)
+@ExtendWith(RequiresDockerCondition.class)
+@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class ConfigDrivenIndexingIntegrationTest
{
@@ -98,7 +100,7 @@ class ConfigDrivenIndexingIntegrationTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
- qInstance.setAuthentication(new QAuthenticationMetaData()
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/IndexDriftIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/IndexDriftIntegrationTest.java
new file mode 100644
index 0000000..2c99aad
--- /dev/null
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/IndexDriftIntegrationTest.java
@@ -0,0 +1,283 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch;
+
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchAction;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchInput;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchResult;
+import com.kingsrook.qbits.quicksearch.annotations.QuickSearchField;
+import com.kingsrook.qbits.quicksearch.annotations.QuickSearchable;
+import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
+import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
+import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Integration tests, against a real OpenSearch, that real-time indexing keeps
+ ** the index in step with the source table through updates and deletes, and
+ ** that the reconcile process repairs an index that has drifted.
+ **
+ ** Each test uses its own record IDs and search terms, so tests are independent.
+ *******************************************************************************/
+@ExtendWith(RequiresDockerCondition.class)
+@Testcontainers
+class IndexDriftIntegrationTest
+{
+ private static final String BACKEND_NAME = "driftMemoryBackend";
+ private static final String TABLE_NAME = "driftCustomer";
+ private static final String INDEX_NAME = "quick_search_drift_test";
+
+ @Container
+ static GenericContainer opensearch = new GenericContainer<>("opensearchproject/opensearch:2.11.0")
+ .withExposedPorts(9200)
+ .withEnv("discovery.type", "single-node")
+ .withEnv("plugins.security.disabled", "true")
+ .withEnv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "Admin123!")
+ .waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200));
+
+
+
+ /*******************************************************************************
+ ** Entity annotated for quick search indexing.
+ *******************************************************************************/
+ @QuickSearchable(tableName = TABLE_NAME)
+ static class DriftCustomer
+ {
+ @QuickSearchField(weight = 2)
+ private String name;
+
+ @QuickSearchField
+ private String city;
+ }
+
+
+
+ /*******************************************************************************
+ ** Build the instance, produce the QBit with real-time indexing on, and
+ ** initialize QContext.
+ *******************************************************************************/
+ @BeforeAll
+ static void setUpAll() throws Exception
+ {
+ MemoryRecordStore.getInstance().reset();
+
+ QInstance qInstance = new QInstance();
+ qInstance.addBackend(new QBackendMetaData()
+ .withName(BACKEND_NAME)
+ .withBackendType(MemoryBackendModule.class));
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
+ .withName("anonymous")
+ .withType(QAuthenticationType.FULLY_ANONYMOUS));
+ qInstance.addTable(new QTableMetaData()
+ .withName(TABLE_NAME)
+ .withBackendName(BACKEND_NAME)
+ .withPrimaryKeyField("id")
+ .withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
+ .withField(new QFieldMetaData("name", QFieldType.STRING))
+ .withField(new QFieldMetaData("city", QFieldType.STRING)));
+
+ QuickSearchQBitConfig config = new QuickSearchQBitConfig()
+ .withBackendName(BACKEND_NAME)
+ .withOpensearchHost(opensearch.getHost())
+ .withOpensearchPort(opensearch.getMappedPort(9200))
+ .withOpensearchIndexName(INDEX_NAME)
+ .withSearchableEntityClasses(List.of(DriftCustomer.class))
+ .withEnableRealTimeIndexing(true);
+
+ new QuickSearchQBitProducer().withConfig(config).produce(qInstance);
+
+ QContext.init(qInstance, new QSession());
+ }
+
+
+
+ /*******************************************************************************
+ ** Clean up QContext and QBit state after all tests complete.
+ *******************************************************************************/
+ @AfterAll
+ static void tearDownAll()
+ {
+ QContext.clear();
+ QuickSearchQBitContext.clear();
+ MemoryRecordStore.getInstance().reset();
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: a partial update (one field) keeps the other fields searchable.
+ *******************************************************************************/
+ @Test
+ void testPartialUpdate_unchangedFieldsStaySearchable() throws Exception
+ {
+ insert(new QRecord().withValue("id", 1).withValue("name", "Ada").withValue("city", "Lovelaceville"));
+
+ UpdateInput updateInput = new UpdateInput();
+ updateInput.setTableName(TABLE_NAME);
+ updateInput.setRecords(List.of(new QRecord().withValue("id", 1).withValue("name", "Adeline")));
+ new UpdateAction().execute(updateInput);
+
+ assertThat(searchRecordIds("lovelaceville"))
+ .as("the unchanged city must stay searchable after updating only the name")
+ .containsExactly("1");
+ assertThat(searchRecordIds("adeline")).containsExactly("1");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: deleting records removes their documents, even when the batch also
+ ** holds a record whose document is already missing from the index.
+ *******************************************************************************/
+ @Test
+ void testDelete_removesDocuments_evenWithMissingDocumentInBatch() throws Exception
+ {
+ insert(
+ new QRecord().withValue("id", 10).withValue("name", "Tenley").withValue("city", "Deleteton"),
+ new QRecord().withValue("id", 11).withValue("name", "Elvira").withValue("city", "Deleteton"),
+ new QRecord().withValue("id", 12).withValue("name", "Twila").withValue("city", "Deleteton"));
+
+ ////////////////////////////////////////////////////////////////////
+ // simulate earlier drift: record 12's document is already gone //
+ ////////////////////////////////////////////////////////////////////
+ client().deleteDocument(TABLE_NAME, "12");
+
+ DeleteInput deleteInput = new DeleteInput();
+ deleteInput.setTableName(TABLE_NAME);
+ deleteInput.setPrimaryKeys(List.of(12, 10));
+ new DeleteAction().execute(deleteInput);
+
+ assertThat(searchRecordIds("deleteton"))
+ .as("only the record that was not deleted should remain")
+ .containsExactly("11");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: the reconcile process repairs each kind of drift from the source
+ ** table: an orphan document (record gone), a missing document, and a stale
+ ** document (record changed since it was indexed).
+ *******************************************************************************/
+ @Test
+ void testReconcile_repairsOrphanMissingAndStaleDocuments() throws Exception
+ {
+ insert(
+ new QRecord().withValue("id", 21).withValue("name", "Stella").withValue("city", "Freshford"),
+ new QRecord().withValue("id", 22).withValue("name", "Mona").withValue("city", "Missingham"));
+
+ ////////////////////////////////////////////////////////////////////
+ // drift the index away from the source table three ways //
+ ////////////////////////////////////////////////////////////////////
+ Instant anHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ client().indexDocuments(List.of(
+ new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("99")
+ .withSearchableText("Ghostly Phantomburg").withIndexedAt(anHourAgo).withFieldValues(Map.of()),
+ new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("21")
+ .withSearchableText("Stella Oldtown").withIndexedAt(anHourAgo).withFieldValues(Map.of())), 10);
+ client().deleteDocument(TABLE_NAME, "22");
+
+ assertThat(searchRecordIds("phantomburg")).containsExactly("99");
+ assertThat(searchRecordIds("oldtown")).containsExactly("21");
+ assertThat(searchRecordIds("missingham")).isEmpty();
+
+ RunBackendStepInput input = new RunBackendStepInput();
+ input.addValue("tableName", TABLE_NAME);
+ RunBackendStepOutput output = new RunBackendStepOutput();
+ new ReconcileIndexStep().run(input, output);
+
+ assertThat(searchRecordIds("phantomburg")).as("orphan document removed").isEmpty();
+ assertThat(searchRecordIds("oldtown")).as("stale document replaced").isEmpty();
+ assertThat(searchRecordIds("freshford")).as("stale document replaced").containsExactly("21");
+ assertThat(searchRecordIds("missingham")).as("missing document restored").containsExactly("22");
+ assertThat(output.getValue("documentsRemoved")).isEqualTo(1L);
+ }
+
+
+
+ /*******************************************************************************
+ ** Insert records through InsertAction (so real-time indexing runs).
+ *******************************************************************************/
+ private void insert(QRecord... records) throws QException
+ {
+ InsertInput insertInput = new InsertInput();
+ insertInput.setTableName(TABLE_NAME);
+ insertInput.setRecords(List.of(records));
+ new InsertAction().execute(insertInput);
+ }
+
+
+
+ /*******************************************************************************
+ ** Refresh the index, then return the record IDs matching the term.
+ *******************************************************************************/
+ private List searchRecordIds(String term) throws QException
+ {
+ client().refreshIndex();
+ return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME))
+ .getResults().stream()
+ .map(QuickSearchResult::getRecordId)
+ .toList());
+ }
+
+
+
+ /*******************************************************************************
+ ** The live OpenSearch client from the QBit context.
+ *******************************************************************************/
+ private QuickSearchOpenSearchClient client()
+ {
+ return (QuickSearchQBitContext.getClient());
+ }
+
+}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/OpenSearchIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/OpenSearchIntegrationTest.java
index bcfc09e..f9eed5c 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/OpenSearchIntegrationTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/OpenSearchIntegrationTest.java
@@ -47,6 +47,7 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
@@ -61,7 +62,8 @@ import static org.assertj.core.api.Assertions.assertThat;
** Tests run in order: index creation, full reindex, search with results,
** pagination, and no-results search.
*******************************************************************************/
-@Testcontainers(disabledWithoutDocker = true)
+@ExtendWith(RequiresDockerCondition.class)
+@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class OpenSearchIntegrationTest
{
@@ -115,7 +117,7 @@ class OpenSearchIntegrationTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
- qInstance.setAuthentication(new QAuthenticationMetaData()
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducerTest.java
index c3d4d3c..7e629e9 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducerTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducerTest.java
@@ -162,6 +162,7 @@ class QuickSearchQBitProducerTest
assertThat(qInstance.getTable("myApp_quickSearchIndexRun")).isNotNull();
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
+ assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
}
@@ -342,6 +343,7 @@ class QuickSearchQBitProducerTest
assertThat(qInstance.getProcess(QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess(QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
+ assertThat(qInstance.getProcess(QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
}
@@ -794,7 +796,7 @@ class QuickSearchQBitProducerTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
- qInstance.setAuthentication(new QAuthenticationMetaData()
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerCondition.java b/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerCondition.java
new file mode 100644
index 0000000..df8069b
--- /dev/null
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerCondition.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch;
+
+
+import org.junit.jupiter.api.extension.ConditionEvaluationResult;
+import org.junit.jupiter.api.extension.ExecutionCondition;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.testcontainers.DockerClientFactory;
+
+
+/*******************************************************************************
+ ** JUnit condition for integration tests that need Docker.
+ **
+ ** Without Docker, the tests are skipped on a developer machine, but fail when
+ ** CI=true, so a CI build can never pass with its integration tests silently
+ ** skipped. Register it ahead of @Testcontainers, so no container is started
+ ** when the tests are disabled.
+ *******************************************************************************/
+public class RequiresDockerCondition implements ExecutionCondition
+{
+
+ /*******************************************************************************
+ ** Evaluate against the local Docker environment and the CI env var.
+ *******************************************************************************/
+ @Override
+ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context)
+ {
+ return (evaluate(DockerClientFactory.instance().isDockerAvailable(), System.getenv("CI")));
+ }
+
+
+
+ /*******************************************************************************
+ ** Decide whether Docker-backed tests run, skip, or fail.
+ **
+ ** @param dockerAvailable whether a usable Docker environment was found
+ ** @param ciValue value of the CI environment variable (may be null)
+ ** @return enabled when Docker is available; disabled when it is not and CI
+ ** is not "true"
+ ** @throws IllegalStateException when Docker is unavailable and CI is "true"
+ *******************************************************************************/
+ static ConditionEvaluationResult evaluate(Boolean dockerAvailable, String ciValue)
+ {
+ if(Boolean.TRUE.equals(dockerAvailable))
+ {
+ return (ConditionEvaluationResult.enabled("Docker is available"));
+ }
+
+ if("true".equalsIgnoreCase(ciValue))
+ {
+ throw (new IllegalStateException("Docker is required for integration tests when CI=true, but no usable Docker environment was found"));
+ }
+
+ return (ConditionEvaluationResult.disabled("Docker is not available; skipping (with CI=true these tests fail instead)"));
+ }
+
+}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerConditionTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerConditionTest.java
new file mode 100644
index 0000000..cdb6a29
--- /dev/null
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerConditionTest.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch;
+
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ConditionEvaluationResult;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Tests for RequiresDockerCondition.
+ *******************************************************************************/
+class RequiresDockerConditionTest
+{
+
+ /*******************************************************************************
+ ** Docker available: integration tests run, whether or not CI is set.
+ *******************************************************************************/
+ @Test
+ void testEvaluate_dockerAvailable_enabled()
+ {
+ ConditionEvaluationResult local = RequiresDockerCondition.evaluate(true, null);
+ ConditionEvaluationResult ci = RequiresDockerCondition.evaluate(true, "true");
+
+ assertThat(local.isDisabled()).isFalse();
+ assertThat(ci.isDisabled()).isFalse();
+ }
+
+
+
+ /*******************************************************************************
+ ** No Docker outside CI: integration tests are skipped.
+ *******************************************************************************/
+ @Test
+ void testEvaluate_noDockerOutsideCi_disabled()
+ {
+ assertThat(RequiresDockerCondition.evaluate(false, null).isDisabled()).isTrue();
+ assertThat(RequiresDockerCondition.evaluate(false, "").isDisabled()).isTrue();
+ assertThat(RequiresDockerCondition.evaluate(false, "false").isDisabled()).isTrue();
+ }
+
+
+
+ /*******************************************************************************
+ ** No Docker with CI=true: evaluation throws, so the tests fail instead of
+ ** silently skipping.
+ *******************************************************************************/
+ @Test
+ void testEvaluate_noDockerInCi_throws()
+ {
+ assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "true"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("CI=true");
+
+ assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "TRUE"))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java
index b699b3c..8d527b4 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java
@@ -29,6 +29,7 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -79,7 +80,7 @@ class QuickSearchPostDeleteCustomizerTest
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
- qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
QContext.init(qInstance, new QSession());
}
@@ -144,6 +145,50 @@ class QuickSearchPostDeleteCustomizerTest
+ /*******************************************************************************
+ ** Test that records the delete rejected (with errors), or that carry no
+ ** primary key, do not remove documents from the index.
+ *******************************************************************************/
+ @SuppressWarnings("unchecked")
+ @Test
+ void testPostDelete_recordsWithErrorsOrNoPrimaryKey_skipped() throws QException
+ {
+ QRecord deleted = new QRecord().withValue(PK_FIELD, 5);
+ QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
+ rejected.addError(new BadInputStatusMessage("still referenced"));
+ QRecord noKey = new QRecord().withValue("name", "orphan");
+
+ DeleteInput deleteInput = new DeleteInput();
+ deleteInput.setTableName(TABLE_NAME);
+
+ List result = new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(deleted, rejected, noKey));
+ assertThat(result).containsExactly(deleted, rejected, noKey);
+
+ ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class);
+ verify(mockPublisher).publishDeleteEvents(captor.capture());
+ assertThat(captor.getValue()).extracting(IndexEvent::getRecordId).containsExactly("5");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test that when no record is eligible, nothing is published.
+ *******************************************************************************/
+ @Test
+ void testPostDelete_onlyRejectedRecords_nothingPublished() throws QException
+ {
+ QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
+ rejected.addError(new BadInputStatusMessage("still referenced"));
+
+ DeleteInput deleteInput = new DeleteInput();
+ deleteInput.setTableName(TABLE_NAME);
+
+ new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(rejected));
+ verify(mockPublisher, never()).publishDeleteEvents(anyList());
+ }
+
+
+
/*******************************************************************************
** Test that post-delete with empty records list returns input without calling
** publisher.
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java
index 7e2d8e7..5fd2406 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java
@@ -79,7 +79,7 @@ class QuickSearchPostInsertCustomizerTest
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
- qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
QContext.init(qInstance, new QSession());
}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java
index b9c6c86..d11e772 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java
@@ -19,17 +19,26 @@ package com.kingsrook.qbits.quicksearch.customizers;
import java.util.List;
import java.util.Optional;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
-import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
-import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -50,23 +59,29 @@ import static org.mockito.Mockito.verify;
/*******************************************************************************
** Tests for QuickSearchPostUpdateCustomizer.
+ **
+ ** Uses a memory backend, because the customizer re-reads updated records from
+ ** the source table before indexing them.
*******************************************************************************/
class QuickSearchPostUpdateCustomizerTest
{
- private static final String TABLE_NAME = "orders";
- private static final String PK_FIELD = "id";
+ private static final String BACKEND_NAME = "memoryBackend";
+ private static final String TABLE_NAME = "orders";
+ private static final String PK_FIELD = "id";
private IndexEventPublisher mockPublisher;
/*******************************************************************************
- ** Set up mock publisher in QuickSearchQBitContext and a QContext with a test
- ** QInstance before each test.
+ ** Set up mock publisher in QuickSearchQBitContext and a QContext with a
+ ** memory-backed test QInstance before each test.
*******************************************************************************/
@BeforeEach
void setUp()
{
+ MemoryRecordStore.getInstance().reset();
+
mockPublisher = mock(IndexEventPublisher.class);
QuickSearchQBitContext.setPublisher(mockPublisher);
QuickSearchQBitContext.setDiscoveredTables(List.of(
@@ -74,14 +89,19 @@ class QuickSearchPostUpdateCustomizerTest
));
QInstance qInstance = new QInstance();
+ qInstance.addBackend(new QBackendMetaData()
+ .withName(BACKEND_NAME)
+ .withBackendType(MemoryBackendModule.class));
qInstance.addTable(new QTableMetaData()
.withName(TABLE_NAME)
+ .withBackendName(BACKEND_NAME)
.withPrimaryKeyField(PK_FIELD)
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
- .withField(new QFieldMetaData("name", QFieldType.STRING)));
- qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
- qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
- qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
+ .withField(new QFieldMetaData("name", QFieldType.STRING))
+ .withField(new QFieldMetaData("description", QFieldType.STRING)));
+ qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
+ .withName("anonymous")
+ .withType(QAuthenticationType.FULLY_ANONYMOUS));
QContext.init(qInstance, new QSession());
}
@@ -95,6 +115,33 @@ class QuickSearchPostUpdateCustomizerTest
{
QuickSearchQBitContext.clear();
QContext.clear();
+ MemoryRecordStore.getInstance().reset();
+ }
+
+
+
+ /*******************************************************************************
+ ** Insert full records into the source table.
+ *******************************************************************************/
+ private void insertSourceRecords(QRecord... records) throws QException
+ {
+ InsertInput insertInput = new InsertInput();
+ insertInput.setTableName(TABLE_NAME);
+ insertInput.setRecords(List.of(records));
+ new InsertAction().execute(insertInput);
+ }
+
+
+
+ /*******************************************************************************
+ ** Capture the events passed to the publisher's publishIndexEvents.
+ *******************************************************************************/
+ @SuppressWarnings("unchecked")
+ private List captureIndexEvents() throws QException
+ {
+ ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class);
+ verify(mockPublisher).publishIndexEvents(captor.capture());
+ return (captor.getValue());
}
@@ -111,36 +158,107 @@ class QuickSearchPostUpdateCustomizerTest
/*******************************************************************************
- ** Test that post-update with valid records builds correct INDEX events and
- ** calls publisher.publishIndexEvents.
+ ** Test that post-update builds INDEX events from the stored source records,
+ ** not from the (possibly sparse) records passed in by the update.
*******************************************************************************/
- @SuppressWarnings("unchecked")
@Test
- void testPostUpdate_withRecords_publishesIndexEvents() throws QException
+ void testPostUpdate_withRecords_indexesStoredSourceRecords() throws QException
{
- QRecord record1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
- QRecord record2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
+ insertSourceRecords(
+ new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated").withValue("description", "Engineer"),
+ new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated").withValue("description", "Designer"));
+
+ QRecord sparse1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
+ QRecord sparse2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
- List result = customizer.postUpdate(updateInput, List.of(record1, record2), Optional.empty());
+ List result = customizer.postUpdate(updateInput, List.of(sparse1, sparse2), Optional.empty());
- assertThat(result).containsExactly(record1, record2);
+ assertThat(result).containsExactly(sparse1, sparse2);
- ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class);
- verify(mockPublisher).publishIndexEvents(captor.capture());
-
- List events = captor.getValue();
+ List events = captureIndexEvents();
assertThat(events).hasSize(2);
- assertThat(events.get(0).getAction()).isEqualTo(IndexEventAction.INDEX);
- assertThat(events.get(0).getTableName()).isEqualTo(TABLE_NAME);
- assertThat(events.get(0).getRecordId()).isEqualTo("10");
- assertThat(events.get(0).getRecord()).isSameAs(record1);
- assertThat(events.get(1).getAction()).isEqualTo(IndexEventAction.INDEX);
- assertThat(events.get(1).getRecordId()).isEqualTo("20");
- assertThat(events.get(1).getRecord()).isSameAs(record2);
+ assertThat(events).allSatisfy(e ->
+ {
+ assertThat(e.getAction()).isEqualTo(IndexEventAction.INDEX);
+ assertThat(e.getTableName()).isEqualTo(TABLE_NAME);
+ });
+ assertThat(events).extracting(IndexEvent::getRecordId).containsExactlyInAnyOrder("10", "20");
+
+ IndexEvent aliceEvent = events.stream().filter(e -> "10".equals(e.getRecordId())).findFirst().orElseThrow();
+ assertThat(aliceEvent.getRecord().getValueString("name")).isEqualTo("Alice Updated");
+ assertThat(aliceEvent.getRecord().getValueString("description")).isEqualTo("Engineer");
+ }
+
+
+
+ /*******************************************************************************
+ ** End to end through UpdateAction: a partial update of one field must not drop
+ ** the other indexed fields from the document.
+ *******************************************************************************/
+ @Test
+ void testUpdateAction_partialUpdate_reindexesFullRecord() throws QException
+ {
+ QContext.getQInstance().getTable(TABLE_NAME)
+ .withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(QuickSearchPostUpdateCustomizer.class));
+
+ insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice").withValue("description", "Engineer"));
+
+ UpdateInput updateInput = new UpdateInput();
+ updateInput.setTableName(TABLE_NAME);
+ updateInput.setRecords(List.of(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alicia")));
+ new UpdateAction().execute(updateInput);
+
+ List events = captureIndexEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0).getRecordId()).isEqualTo("1");
+ assertThat(events.get(0).getRecord().getValueString("name")).isEqualTo("Alicia");
+ assertThat(events.get(0).getRecord().getValueString("description")).isEqualTo("Engineer");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test that records the update rejected (with errors) are not re-indexed.
+ *******************************************************************************/
+ @Test
+ void testPostUpdate_recordWithErrors_notIndexed() throws QException
+ {
+ insertSourceRecords(
+ new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good"),
+ new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected"));
+
+ QRecord good = new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good");
+ QRecord rejected = new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected");
+ rejected.addError(new BadInputStatusMessage("not allowed"));
+
+ UpdateInput updateInput = new UpdateInput();
+ updateInput.setTableName(TABLE_NAME);
+
+ new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(good, rejected), Optional.empty());
+
+ List events = captureIndexEvents();
+ assertThat(events).extracting(IndexEvent::getRecordId).containsExactly("1");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test that when no updated record can be read back from the source table,
+ ** nothing is published.
+ *******************************************************************************/
+ @Test
+ void testPostUpdate_recordNotInSource_nothingPublished() throws QException
+ {
+ UpdateInput updateInput = new UpdateInput();
+ updateInput.setTableName(TABLE_NAME);
+
+ new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(new QRecord().withValue(PK_FIELD, 99)), Optional.empty());
+
+ verify(mockPublisher, never()).publishIndexEvents(anyList());
}
@@ -209,6 +327,7 @@ class QuickSearchPostUpdateCustomizerTest
@Test
void testPostUpdate_publisherThrows_returnsRecordsWithoutThrowing() throws QException
{
+ insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice"));
doThrow(new QException("publish failed"))
.when(mockPublisher).publishIndexEvents(anyList());
@@ -218,6 +337,7 @@ class QuickSearchPostUpdateCustomizerTest
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
assertThatNoException().isThrownBy(() -> customizer.postUpdate(updateInput, List.of(record), Optional.empty()));
+ verify(mockPublisher).publishIndexEvents(anyList());
}
}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClientTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClientTest.java
index 985b317..4471dfc 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClientTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClientTest.java
@@ -261,4 +261,51 @@ class QuickSearchOpenSearchClientTest
}
}
+
+
+ /*******************************************************************************
+ ** deleteDocuments with a null or empty list is a no-op, even with no server.
+ *******************************************************************************/
+ @Test
+ void testDeleteDocuments_emptyList_returnsEmptyResult() throws QException
+ {
+ QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
+
+ try
+ {
+ BulkIndexResult emptyResult = client.deleteDocuments(List.of(), 100);
+ BulkIndexResult nullResult = client.deleteDocuments(null, 100);
+
+ assertThat(emptyResult.getSuccessCount()).isEqualTo(0);
+ assertThat(emptyResult.getFailureCount()).isEqualTo(0);
+ assertThat(nullResult.getSuccessCount()).isEqualTo(0);
+ }
+ finally
+ {
+ client.close();
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** deleteDocuments throws QException when the server is unavailable.
+ *******************************************************************************/
+ @Test
+ void testDeleteDocuments_serverUnavailable_throwsQException() throws QException
+ {
+ QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
+
+ try
+ {
+ assertThatThrownBy(() -> client.deleteDocuments(List.of("orders:1", "orders:2"), 100))
+ .isInstanceOf(QException.class)
+ .hasMessageContaining("Bulk delete");
+ }
+ finally
+ {
+ client.close();
+ }
+ }
+
}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java
new file mode 100644
index 0000000..b371d1d
--- /dev/null
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java
@@ -0,0 +1,281 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch.processes;
+
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
+import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
+import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
+import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
+import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+/*******************************************************************************
+ ** Tests for ReconcileIndexStep.
+ *******************************************************************************/
+class ReconcileIndexStepTest extends BaseQuickSearchTest
+{
+ private QuickSearchOpenSearchClient mockClient;
+
+
+
+ /*******************************************************************************
+ ** Install a mock client that reports every document as indexed.
+ *******************************************************************************/
+ @BeforeEach
+ void setUpClient() throws QException
+ {
+ mockClient = mock(QuickSearchOpenSearchClient.class);
+ when(mockClient.indexDocuments(anyList(), anyInt())).thenAnswer(invocation ->
+ {
+ List docs = invocation.getArgument(0);
+ return (new BulkIndexResult().withSuccessCount(docs.size()));
+ });
+ when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(0L);
+ QuickSearchQBitContext.setClient(mockClient);
+ }
+
+
+
+ /*******************************************************************************
+ ** Insert count source records into the testEntity table.
+ *******************************************************************************/
+ private void insertTestEntities(int count) throws QException
+ {
+ List records = new ArrayList<>();
+ for(int i = 1; i <= count; i++)
+ {
+ records.add(new QRecord().withValue("name", "Entity " + i).withValue("description", "Description " + i));
+ }
+
+ InsertInput insertInput = new InsertInput();
+ insertInput.setTableName(TEST_ENTITY_TABLE);
+ insertInput.setRecords(records);
+ new InsertAction().execute(insertInput);
+ }
+
+
+
+ /*******************************************************************************
+ ** Query all rows of the given table.
+ *******************************************************************************/
+ private List queryAll(String tableName) throws QException
+ {
+ QueryInput queryInput = new QueryInput();
+ queryInput.setTableName(tableName);
+ queryInput.setFilter(new QQueryFilter());
+ return (new QueryAction().execute(queryInput).getRecords());
+ }
+
+
+
+ /*******************************************************************************
+ ** Collect the record IDs of every document passed to indexDocuments.
+ *******************************************************************************/
+ @SuppressWarnings("unchecked")
+ private List indexedRecordIds() throws QException
+ {
+ ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class);
+ verify(mockClient, atLeastOnce()).indexDocuments(captor.capture(), anyInt());
+
+ List recordIds = new ArrayList<>();
+ for(List batch : captor.getAllValues())
+ {
+ batch.forEach(doc -> recordIds.add(doc.getRecordId()));
+ }
+ return (recordIds);
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: every source record is re-indexed, then documents not re-indexed by
+ ** this run are removed, after a refresh. The table is never wiped first.
+ *******************************************************************************/
+ @Test
+ void testReconcile_reindexesSourceThenRemovesStaleDocuments() throws QException
+ {
+ insertTestEntities(3);
+ when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(2L);
+
+ Instant beforeRun = Instant.now();
+ RunBackendStepOutput output = new RunBackendStepOutput();
+ new ReconcileIndexStep().run(new RunBackendStepInput(), output);
+ Instant afterRun = Instant.now();
+
+ assertThat(indexedRecordIds()).containsExactlyInAnyOrder("1", "2", "3");
+
+ ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(Instant.class);
+ InOrder inOrder = inOrder(mockClient);
+ inOrder.verify(mockClient).indexDocuments(anyList(), anyInt());
+ inOrder.verify(mockClient).refreshIndex();
+ inOrder.verify(mockClient).deleteDocumentsIndexedBefore(eq(TEST_ENTITY_TABLE), cutoffCaptor.capture());
+
+ assertThat(cutoffCaptor.getValue()).isBetween(beforeRun, afterRun);
+ verify(mockClient, never()).deleteDocumentsForTable(anyString());
+
+ assertThat(output.getValueInteger("recordsIndexed")).isEqualTo(3);
+ assertThat(output.getValue("documentsRemoved")).isEqualTo(2L);
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: source records are paged by primary key, so every record is read
+ ** exactly once across batches.
+ *******************************************************************************/
+ @Test
+ void testReconcile_pagesByPrimaryKey_eachRecordOnce() throws QException
+ {
+ QuickSearchQBitContext.getConfig().withSourceBatchSize(2);
+ insertTestEntities(5);
+
+ new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+ verify(mockClient, times(3)).indexDocuments(anyList(), anyInt());
+ assertThat(indexedRecordIds()).containsExactly("1", "2", "3", "4", "5");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: when any document fails to index, stale documents are not removed
+ ** (a document that failed to re-index would otherwise be purged), and the
+ ** run is marked FAILED.
+ *******************************************************************************/
+ @Test
+ void testReconcile_indexingErrors_keepsDocumentsAndFailsRun() throws QException
+ {
+ insertTestEntities(2);
+ BulkIndexResult partial = new BulkIndexResult();
+ partial.addSuccess();
+ partial.addFailure("testEntity:2: mapper_parsing_exception");
+ when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(partial);
+
+ new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+ verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
+
+ List runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
+ assertThat(runs).hasSize(1);
+ assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
+ assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
+ assertThat(runs.get(0).getValueInteger("errorCount")).isEqualTo(1);
+ assertThat(runs.get(0).getValueString("errorMessage")).contains("mapper_parsing_exception");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: a client exception fails the run record and is rethrown.
+ *******************************************************************************/
+ @Test
+ void testReconcile_clientThrows_failsRunAndThrows() throws QException
+ {
+ insertTestEntities(1);
+ when(mockClient.indexDocuments(anyList(), anyInt())).thenThrow(new QException("connection refused"));
+
+ assertThatThrownBy(() -> new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
+ .isInstanceOf(QException.class)
+ .hasMessageContaining("Reconcile failed");
+
+ List runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
+ assertThat(runs).hasSize(1);
+ assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
+ assertThat(runs.get(0).getValueString("errorMessage")).contains("connection refused");
+ verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: a successful run completes its run record and updates the index row.
+ *******************************************************************************/
+ @Test
+ void testReconcile_success_completesRunAndUpdatesIndexRow() throws QException
+ {
+ insertTestEntities(4);
+
+ Instant beforeRun = Instant.now();
+ new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+ List runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
+ assertThat(runs).hasSize(1);
+ assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
+ assertThat(runs.get(0).getValueString("status")).isEqualTo("COMPLETED");
+ assertThat(runs.get(0).getValueInteger("recordsProcessed")).isEqualTo(4);
+ assertThat(runs.get(0).getValueInteger("recordsIndexed")).isEqualTo(4);
+
+ List indexRows = queryAll(QuickSearchIndex.TABLE_NAME);
+ assertThat(indexRows).hasSize(1);
+ assertThat(indexRows.get(0).getValueInteger("recordCount")).isEqualTo(4);
+ assertThat(indexRows.get(0).getValueInstant("lastFullReindexTime")).isAfterOrEqualTo(beforeRun);
+ assertThat(indexRows.get(0).getValueInstant("lastBasepullTime")).isAfterOrEqualTo(beforeRun);
+ }
+
+
+
+ /*******************************************************************************
+ ** Test: the tableName input limits the run to that table.
+ *******************************************************************************/
+ @Test
+ void testReconcile_unknownTableNameInput_nothingReconciled() throws QException
+ {
+ insertTestEntities(1);
+
+ RunBackendStepInput input = new RunBackendStepInput();
+ input.addValue("tableName", "someOtherTable");
+ new ReconcileIndexStep().run(input, new RunBackendStepOutput());
+
+ verify(mockClient, never()).indexDocuments(anyList(), anyInt());
+ verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
+ assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).isEmpty();
+ }
+
+}
diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisherTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisherTest.java
index 450636c..b0e9208 100644
--- a/src/test/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisherTest.java
+++ b/src/test/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisherTest.java
@@ -34,6 +34,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -167,10 +168,11 @@ class SynchronousIndexEventPublisherTest
/*******************************************************************************
- ** Test publishDeleteEvents calls client.deleteDocument for each event.
+ ** Test publishDeleteEvents removes all documents in one bulk call, so one
+ ** failing document cannot stop the others from being removed.
*******************************************************************************/
@Test
- void testPublishDeleteEvents_callsClientForEachEvent() throws QException
+ void testPublishDeleteEvents_bulkDeletesAllDocuments() throws QException
{
List events = List.of(
new IndexEvent()
@@ -182,10 +184,61 @@ class SynchronousIndexEventPublisherTest
.withRecordId("20")
.withAction(IndexEventAction.DELETE));
+ when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult());
+
publisher.publishDeleteEvents(events);
- verify(mockClient).deleteDocument("orders", "10");
- verify(mockClient).deleteDocument("orders", "20");
+ verify(mockClient).deleteDocuments(List.of("orders:10", "orders:20"), BATCH_SIZE);
+ verify(mockClient, never()).deleteDocument(anyString(), anyString());
+ }
+
+
+
+ /*******************************************************************************
+ ** Test publishDeleteEvents throws, naming the failed documents, when some
+ ** bulk items fail.
+ *******************************************************************************/
+ @Test
+ void testPublishDeleteEvents_itemFailures_throwsQException() throws QException
+ {
+ BulkIndexResult result = new BulkIndexResult();
+ result.addSuccess();
+ result.addFailure("orders:20: shard unavailable");
+ when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(result);
+
+ List events = List.of(
+ new IndexEvent().withTableName("orders").withRecordId("10").withAction(IndexEventAction.DELETE),
+ new IndexEvent().withTableName("orders").withRecordId("20").withAction(IndexEventAction.DELETE));
+
+ assertThatThrownBy(() -> publisher.publishDeleteEvents(events))
+ .isInstanceOf(QException.class)
+ .hasMessageContaining("1 of 2")
+ .hasMessageContaining("orders:20: shard unavailable");
+ }
+
+
+
+ /*******************************************************************************
+ ** Test publishIndexEvents throws, naming the failed documents, when some
+ ** bulk items fail (instead of silently dropping the failures).
+ *******************************************************************************/
+ @Test
+ void testPublishIndexEvents_itemFailures_throwsQException() throws QException
+ {
+ BulkIndexResult result = new BulkIndexResult();
+ result.addFailure("orders:1: mapper_parsing_exception");
+ when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(result);
+
+ IndexEvent event = new IndexEvent()
+ .withTableName("orders")
+ .withRecordId("1")
+ .withAction(IndexEventAction.INDEX)
+ .withRecord(new QRecord().withValue("id", 1).withValue("orderNumber", "ORD-001"));
+
+ assertThatThrownBy(() -> publisher.publishIndexEvents(List.of(event)))
+ .isInstanceOf(QException.class)
+ .hasMessageContaining("1 of 1")
+ .hasMessageContaining("orders:1: mapper_parsing_exception");
}
@@ -209,7 +262,7 @@ class SynchronousIndexEventPublisherTest
void testPublishDeleteEvents_clientThrows_propagatesException() throws QException
{
doThrow(new QException("delete failed"))
- .when(mockClient).deleteDocument("orders", "1");
+ .when(mockClient).deleteDocuments(anyList(), anyInt());
IndexEvent event = new IndexEvent()
.withTableName("orders")