fix(#24): recover a stuck REBUILDING status after a killed reindex #25
No files matched your search
@@ -140,9 +140,9 @@ QQQ core has its own record search (`RecordSearchAction`, served at `POST /qqq/v
|
||||
|
||||
| Process | Trigger | What it does |
|
||||
|---|---|---|
|
||||
| Quick Search Basepull Index | Scheduled every `basepullRepeatSeconds` when `schedulerName` is set; also on demand | Replays failed real-time events, re-indexes rows changed since the last run start minus `basepullOverlapSeconds`, purges old run history |
|
||||
| Quick Search Basepull Index | Scheduled every `basepullRepeatSeconds` when `schedulerName` is set; also on demand | Recovers a full reindex that stopped without finishing, replays failed real-time events, re-indexes rows changed since the last run start minus `basepullOverlapSeconds`, purges old run history |
|
||||
| Quick Search Reconcile Index | Cron via `reconcileCronExpression` when `schedulerName` is set; also on demand (optional `tableName` input) | Re-indexes everything in place, removes documents without a source row and documents of unconfigured tables |
|
||||
| Quick Search Full Reindex | On demand (optional `tableName` input) | All tables: build a fresh physical index and swap the alias. One table: reconcile algorithm in place |
|
||||
| Quick Search Full Reindex | On demand (optional `tableName` input) | All tables: build a fresh physical index and swap the alias. One table: reconcile algorithm in place. Fails to start while another full reindex is running |
|
||||
|
||||
Without `schedulerName` the processes exist but nothing runs them; the QBit logs a warning at startup.
|
||||
|
||||
@@ -181,6 +181,7 @@ config.withSchedulerName("quartz")
|
||||
| `basepullOverlapSeconds` | 300 | Re-read window before the previous run start |
|
||||
| `defaultBasepullIntervalMinutes` | 5 | Per-table due interval for config-driven tables |
|
||||
| `runHistoryRetentionDays` | 30 | Purge run records older than this; null keeps them |
|
||||
| `fullReindexStaleMinutes` | 30 | A running full reindex that shows no sign of life for this long is treated as stopped: it no longer blocks a new one, and basepull recovers its tables. Keep it well above the time one page of `sourceBatchSize` records takes; null means the default |
|
||||
| `bulkBatchSize`, `maxBulkRequestBytes`, `sourceBatchSize` | 500, 5 MiB, 1000 | Batch sizes |
|
||||
| `maxFieldLength` | 10000 | Truncate each indexed value |
|
||||
| `maxSearchLimit` | 100 | Largest page size |
|
||||
@@ -199,7 +200,7 @@ config.withSchedulerName("quartz")
|
||||
|
||||
- `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE`, `NEEDS_REINDEX` when the configured fields changed since the row was created, or `REBUILDING` while a full reindex of all tables runs). Unique on `tableName`.
|
||||
- `quickSearchIndexRun`: `runType` (`BASEPULL`, `RECONCILE`, `FULL_REINDEX`), `status`, counts, `errorMessage`.
|
||||
- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). While a table is `REBUILDING`, drift detection is suspended for it. If a full reindex is killed mid-run, its tables stay `REBUILDING` and deletes keep collecting as `AWAITING_REINDEX`; no data is lost, and the recovery is to run a full reindex of all tables again, which applies the collected deletes and returns the tables to `ACTIVE`.
|
||||
- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). While a table is `REBUILDING`, drift detection is suspended for it. Only one full reindex runs at a time: starting one while another is running fails. A full reindex shows it is running through its `FULL_REINDEX` run records, which stay `RUNNING` with `modifyDate` refreshed after every page. If it is killed mid-run (pod restart, OOM, deploy), recovery is automatic once its run records have shown no sign of life for `fullReindexStaleMinutes` (30 by default): the next basepull, or the next full reindex, marks those run records `FAILED`, returns the tables to `ACTIVE` with `lastBasepullTime` moved back to the run's start, and turns the `AWAITING_REINDEX` rows `PENDING` so basepull replays the deletes. No data is lost. A run that was only slow and is judged stopped notices before its alias swap and fails without swapping, or, after its swap, stops applying its captured deletes; basepull skips a replayed delete whose record exists again.
|
||||
|
||||
## Extending
|
||||
|
||||
|
||||
@@ -72,6 +72,15 @@ public class QuickSearchQBitConfig implements QBitConfig
|
||||
public static final String DEFAULT_BASEPULL_TIMESTAMP_FIELD = "modifyDate";
|
||||
public static final int DEFAULT_RECORD_SEARCH_TIMEOUT_MILLIS = 2500;
|
||||
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
// a live full reindex refreshes its run record after every page; a page is //
|
||||
// one source query plus sourceBatchSize / bulkBatchSize bulk requests, each //
|
||||
// bounded by the transport timeouts (a few minutes at most by default), //
|
||||
// so 30 minutes leaves a wide margin before a slow run is judged dead, and //
|
||||
// basepull (every 5 minutes by default) recovers a dead one soon after //
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
public static final Integer DEFAULT_FULL_REINDEX_STALE_MINUTES = 30;
|
||||
|
||||
private static final Pattern INDEX_NAME_PATTERN = Pattern.compile("^[a-z0-9][a-z0-9_+.-]*$");
|
||||
private static final Set<String> AWS_SERVICE_NAMES = Set.of("es", "aoss");
|
||||
|
||||
@@ -106,6 +115,7 @@ public class QuickSearchQBitConfig implements QBitConfig
|
||||
private Integer basepullOverlapSeconds = 300;
|
||||
private Integer defaultBasepullIntervalMinutes = 5;
|
||||
private Integer runHistoryRetentionDays = 30;
|
||||
private Integer fullReindexStaleMinutes = DEFAULT_FULL_REINDEX_STALE_MINUTES;
|
||||
private Integer bulkBatchSize = 500;
|
||||
private Integer sourceBatchSize = 1000;
|
||||
private Integer maxFieldLength = 10000;
|
||||
@@ -403,6 +413,7 @@ public class QuickSearchQBitConfig implements QBitConfig
|
||||
requirePositive(recordSearchTimeoutMillis, "recordSearchTimeoutMillis", errors);
|
||||
requirePositive(maxFieldLength, "maxFieldLength", errors);
|
||||
requirePositive(runHistoryRetentionDays, "runHistoryRetentionDays", errors);
|
||||
requirePositive(fullReindexStaleMinutes, "fullReindexStaleMinutes", errors);
|
||||
|
||||
if(basepullOverlapSeconds != null && basepullOverlapSeconds < 0)
|
||||
{
|
||||
@@ -607,6 +618,17 @@ public class QuickSearchQBitConfig implements QBitConfig
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** fullReindexStaleMinutes, defaulting to DEFAULT_FULL_REINDEX_STALE_MINUTES.
|
||||
*******************************************************************************/
|
||||
@JsonIgnore
|
||||
public Integer getEffectiveFullReindexStaleMinutes()
|
||||
{
|
||||
return (fullReindexStaleMinutes == null ? DEFAULT_FULL_REINDEX_STALE_MINUTES : fullReindexStaleMinutes);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@@ -1790,6 +1812,36 @@ public class QuickSearchQBitConfig implements QBitConfig
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for fullReindexStaleMinutes: a RUNNING full reindex whose run record shows no sign of life (start or heartbeat) for this many minutes is treated as stopped; null means the default
|
||||
*******************************************************************************/
|
||||
public Integer getFullReindexStaleMinutes()
|
||||
{
|
||||
return (this.fullReindexStaleMinutes);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for fullReindexStaleMinutes
|
||||
*******************************************************************************/
|
||||
public void setFullReindexStaleMinutes(Integer fullReindexStaleMinutes)
|
||||
{
|
||||
this.fullReindexStaleMinutes = fullReindexStaleMinutes;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for fullReindexStaleMinutes
|
||||
*******************************************************************************/
|
||||
public QuickSearchQBitConfig withFullReindexStaleMinutes(Integer fullReindexStaleMinutes)
|
||||
{
|
||||
this.fullReindexStaleMinutes = fullReindexStaleMinutes;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for bulkBatchSize: documents per OpenSearch bulk request
|
||||
|
||||
@@ -23,8 +23,10 @@ import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
@@ -376,6 +378,64 @@ public abstract class AbstractIndexingStep implements BackendStep
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Refresh a FULL_REINDEX run record's modifyDate: the heartbeat by which
|
||||
** other nodes tell a full reindex that is still going from one that
|
||||
** stopped (see FullReindexStep.recoverDeadRun). Nothing reads it for other
|
||||
** run types. Best effort: a failure is logged and never stops the run (one
|
||||
** judged stopped as a result fails safely before its swap).
|
||||
*******************************************************************************/
|
||||
protected void heartbeat(QuickSearchIndexRun run)
|
||||
{
|
||||
if(run == null || run.getId() == null || !FullReindexStep.RUN_TYPE.equals(run.getRunType()))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName());
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", run.getId()).withValue("modifyDate", Instant.now())));
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
LOG.warn("Could not refresh the full reindex heartbeat", e, logPair("runId", run.getId()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The record IDs, among the rows' recordIds, that exist in the source table.
|
||||
*******************************************************************************/
|
||||
Set<String> queryExistingRecordIds(String tableName, List<QRecord> rows) throws QException
|
||||
{
|
||||
QuickSearchableTableConfig tableConfig = getTableConfig(tableName);
|
||||
if(tableConfig == null)
|
||||
{
|
||||
return (Set.of());
|
||||
}
|
||||
|
||||
String primaryKeyField = tableConfig.getPrimaryKeyField() == null ? "id" : tableConfig.getPrimaryKeyField();
|
||||
List<Serializable> recordIds = new ArrayList<>();
|
||||
rows.forEach(row -> recordIds.add(row.getValueString("recordId")));
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(tableName);
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, recordIds)));
|
||||
|
||||
Set<String> existingIds = new HashSet<>();
|
||||
for(QRecord record : CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()))
|
||||
{
|
||||
existingIds.add(record.getValueString(primaryKeyField));
|
||||
}
|
||||
return (existingIds);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Update fields on a quickSearchIndex row.
|
||||
*******************************************************************************/
|
||||
@@ -429,9 +489,10 @@ public abstract class AbstractIndexingStep implements BackendStep
|
||||
/*******************************************************************************
|
||||
** Index every source record matching the criteria into targetIndex (null
|
||||
** for the alias), paging by primary key so each record is read once even
|
||||
** while rows are inserted or deleted.
|
||||
** while rows are inserted or deleted. The run's heartbeat is refreshed
|
||||
** after every page.
|
||||
*******************************************************************************/
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex) throws QException
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
QuickSearchQBitConfig config = getConfig();
|
||||
String tableName = tableConfig.getTableName();
|
||||
@@ -491,6 +552,8 @@ public abstract class AbstractIndexingStep implements BackendStep
|
||||
{
|
||||
throw new QException("Source record has no value in primary key field [" + primaryKeyField + "]");
|
||||
}
|
||||
|
||||
heartbeat(run);
|
||||
}
|
||||
|
||||
return (new IndexCounts(processed, indexed, errors, skipped, firstError));
|
||||
|
||||
@@ -25,6 +25,7 @@ import java.util.HashMap;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
@@ -55,8 +56,10 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
/*******************************************************************************
|
||||
** Scheduled catch-up. For each enabled table that is due, re-indexes records
|
||||
** whose basepull timestamp is at or after the previous run's start minus the
|
||||
** configured overlap, paging by primary key. First replays failed real-time
|
||||
** events, and afterwards purges old run history.
|
||||
** configured overlap, paging by primary key. First recovers what a full
|
||||
** reindex that stopped without finishing left behind (so its captured deletes
|
||||
** are replayed in the same run) and replays failed real-time events, and
|
||||
** afterwards purges old run history.
|
||||
**
|
||||
** The watermark stored on the quickSearchIndex row is the run's start time,
|
||||
** captured before the first query, so rows modified during the run are read
|
||||
@@ -83,6 +86,16 @@ public class BasepullIndexStep extends AbstractIndexingStep
|
||||
{
|
||||
List<String> failures = new ArrayList<>();
|
||||
|
||||
try
|
||||
{
|
||||
new FullReindexStep().recoverDeadRun(true);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
LOG.warn("Recovering from a stopped full reindex did not complete", e);
|
||||
failures.add("full reindex recovery: " + e.getMessage());
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
replayFailedEvents(output);
|
||||
@@ -211,7 +224,7 @@ public class BasepullIndexStep extends AbstractIndexingStep
|
||||
criteria.add(new QFilterCriteria(timestampField, QCriteriaOperator.GREATER_THAN_OR_EQUALS, since));
|
||||
}
|
||||
|
||||
IndexCounts counts = indexAllRecords(client, tableConfig, criteria, null);
|
||||
IndexCounts counts = indexAllRecords(client, tableConfig, criteria, null, run);
|
||||
|
||||
if(counts.errors() == 0)
|
||||
{
|
||||
@@ -256,8 +269,10 @@ public class BasepullIndexStep extends AbstractIndexingStep
|
||||
|
||||
/*******************************************************************************
|
||||
** Replay PENDING failed real-time events: re-read and index records for
|
||||
** INDEX events, delete documents for DELETE events. Successes are removed;
|
||||
** failures bump attempts and become EXHAUSTED at the limit.
|
||||
** INDEX events, delete documents for DELETE events. A DELETE whose record
|
||||
** exists again (re-created with the same key since) keeps its document and
|
||||
** counts as replayed. Successes are removed; failures bump attempts and
|
||||
** become EXHAUSTED at the limit.
|
||||
*******************************************************************************/
|
||||
void replayFailedEvents(RunBackendStepOutput output) throws QException
|
||||
{
|
||||
@@ -292,19 +307,37 @@ public class BasepullIndexStep extends AbstractIndexingStep
|
||||
|
||||
for(Map.Entry<String, List<QRecord>> entry : deleteByTable.entrySet())
|
||||
{
|
||||
List<String> documentIds = new ArrayList<>();
|
||||
for(QRecord event : entry.getValue())
|
||||
{
|
||||
documentIds.add(OpenSearchDocument.buildDocumentId(entry.getKey(), event.getValueString("recordId")));
|
||||
}
|
||||
List<QRecord> toDelete = entry.getValue();
|
||||
try
|
||||
{
|
||||
BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize());
|
||||
markReplayOutcome(entry.getValue(), result.getFailureCount() == 0, result.getErrors(), succeeded, failed);
|
||||
Set<String> existingIds = queryExistingRecordIds(entry.getKey(), entry.getValue());
|
||||
toDelete = new ArrayList<>();
|
||||
for(QRecord event : entry.getValue())
|
||||
{
|
||||
if(existingIds.contains(event.getValueString("recordId")))
|
||||
{
|
||||
succeeded.add(event.getValue("id"));
|
||||
}
|
||||
else
|
||||
{
|
||||
toDelete.add(event);
|
||||
}
|
||||
}
|
||||
|
||||
if(!toDelete.isEmpty())
|
||||
{
|
||||
List<String> documentIds = new ArrayList<>();
|
||||
for(QRecord event : toDelete)
|
||||
{
|
||||
documentIds.add(OpenSearchDocument.buildDocumentId(entry.getKey(), event.getValueString("recordId")));
|
||||
}
|
||||
BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize());
|
||||
markReplayOutcome(toDelete, result.getFailureCount() == 0, result.getErrors(), succeeded, failed);
|
||||
}
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
markReplayOutcome(entry.getValue(), false, List.of(e.getMessage() == null ? e.toString() : e.getMessage()), succeeded, failed);
|
||||
markReplayOutcome(toDelete, false, List.of(e.getMessage() == null ? e.toString() : e.getMessage()), succeeded, failed);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -19,12 +19,15 @@ package com.kingsrook.qbits.quicksearch.processes;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
@@ -71,6 +74,14 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
** AWAITING_REINDEX failed event. Once the rebuild's own writes are done and
|
||||
** the alias is swapped, those deletes are applied through the alias.
|
||||
**
|
||||
** Only one full reindex runs at a time. Each table's run record is RUNNING
|
||||
** before any table is marked REBUILDING, and its modifyDate is refreshed
|
||||
** after every page; a run whose records show no sign of life for
|
||||
** fullReindexStaleMinutes has stopped (killed JVM, lost node). Every full
|
||||
** reindex, and every basepull, first recovers what such a run left behind
|
||||
** (see recoverDeadRun); a full reindex then refuses to start while another
|
||||
** one is alive.
|
||||
**
|
||||
** With a "tableName" input, that table is rebuilt in place with the reconcile
|
||||
** algorithm (index everything, then delete what was not re-indexed).
|
||||
*******************************************************************************/
|
||||
@@ -88,12 +99,18 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
@Override
|
||||
public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
|
||||
{
|
||||
String tableNameFilter = input.getValueString("tableName");
|
||||
String tableNameFilter = input.getValueString("tableName");
|
||||
boolean allTables = tableNameFilter == null || tableNameFilter.isBlank();
|
||||
|
||||
//////////////////////////////////////////////////////////////////////////////
|
||||
// an all-tables rebuild applies the captured deletes itself after its swap //
|
||||
//////////////////////////////////////////////////////////////////////////////
|
||||
refuseIfAlive(recoverDeadRun(!allTables));
|
||||
|
||||
List<QuickSearchableTableConfig> tables = new ArrayList<>();
|
||||
for(QuickSearchableTableConfig tableConfig : getDiscoveredTables())
|
||||
{
|
||||
if(tableNameFilter != null && !tableNameFilter.isBlank() && !tableNameFilter.equals(tableConfig.getTableName()))
|
||||
if(!allTables && !tableNameFilter.equals(tableConfig.getTableName()))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
@@ -108,7 +125,7 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
}
|
||||
}
|
||||
|
||||
if(tableNameFilter != null && !tableNameFilter.isBlank())
|
||||
if(!allTables)
|
||||
{
|
||||
Integer recordsIndexed = 0;
|
||||
for(QuickSearchableTableConfig tableConfig : tables)
|
||||
@@ -134,29 +151,42 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
QuickSearchOpenSearchClient client = getClient();
|
||||
String physicalIndex = client.newPhysicalIndexName();
|
||||
|
||||
client.createPhysicalIndex(physicalIndex);
|
||||
LOG.info("Rebuilding Quick Search index", logPair("physicalIndex", physicalIndex), logPair("tables", tables.size()));
|
||||
|
||||
Map<QuickSearchableTableConfig, QuickSearchIndexRun> runs = new LinkedHashMap<>();
|
||||
Map<String, IndexCounts> counts = new LinkedHashMap<>();
|
||||
Map<String, Instant> startTimes = new LinkedHashMap<>();
|
||||
Map<String, String> priorStatus = new LinkedHashMap<>();
|
||||
Integer indexed = 0;
|
||||
|
||||
boolean created = false;
|
||||
boolean marked = false;
|
||||
boolean swapped = false;
|
||||
|
||||
try
|
||||
{
|
||||
/////////////////////////////////////////////////////////////////////////
|
||||
// the run records exist before any table is REBUILDING, so whoever //
|
||||
// sees a table REBUILDING also sees this run alive; then make sure no //
|
||||
// other run started at the same moment //
|
||||
/////////////////////////////////////////////////////////////////////////
|
||||
for(QuickSearchableTableConfig tableConfig : tables)
|
||||
{
|
||||
runs.put(tableConfig, createRunRecord(queryIndexId(tableConfig.getTableName()), RUN_TYPE));
|
||||
}
|
||||
refuseIfAlive(queryLiveRunsOtherThan(runs.values()));
|
||||
|
||||
client.createPhysicalIndex(physicalIndex);
|
||||
created = true;
|
||||
LOG.info("Rebuilding Quick Search index", logPair("physicalIndex", physicalIndex), logPair("tables", tables.size()));
|
||||
|
||||
marked = true;
|
||||
markRebuilding(tables, priorStatus);
|
||||
|
||||
for(QuickSearchableTableConfig tableConfig : tables)
|
||||
{
|
||||
String tableName = tableConfig.getTableName();
|
||||
QuickSearchIndexRun run = createRunRecord(queryIndexId(tableName), RUN_TYPE);
|
||||
runs.put(tableConfig, run);
|
||||
String tableName = tableConfig.getTableName();
|
||||
startTimes.put(tableName, Instant.now());
|
||||
|
||||
IndexCounts tableCounts = indexAllRecords(client, tableConfig, List.of(), physicalIndex);
|
||||
IndexCounts tableCounts = indexAllRecords(client, tableConfig, List.of(), physicalIndex, runs.get(tableConfig));
|
||||
counts.put(tableName, tableCounts);
|
||||
indexed = indexed + tableCounts.indexed();
|
||||
|
||||
@@ -167,6 +197,8 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
}
|
||||
|
||||
client.refreshIndex(physicalIndex);
|
||||
keepAlive(runs.values());
|
||||
verifyStillOwned(tables, runs.values());
|
||||
client.swapAliasTo(physicalIndex);
|
||||
swapped = true;
|
||||
|
||||
@@ -174,6 +206,7 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
{
|
||||
String tableName = tableConfig.getTableName();
|
||||
IndexCounts tableCounts = counts.get(tableName);
|
||||
keepAlive(runs.values());
|
||||
|
||||
Map<String, Serializable> values = new HashMap<>();
|
||||
values.put("lastFullReindexTime", Instant.now());
|
||||
@@ -188,30 +221,31 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
values.put("status", STATUS_ACTIVE);
|
||||
updateIndexRow(runs.get(tableConfig).getQuickSearchIndexId(), values);
|
||||
priorStatus.remove(tableName);
|
||||
|
||||
completeRunRecord(runs.get(tableConfig), RUN_COMPLETED, tableCounts.processed(), tableCounts.indexed(), 0, null);
|
||||
}
|
||||
|
||||
applyCapturedDeletes(client, tables);
|
||||
/////////////////////////////////////////////////////////////////////////////
|
||||
// the run records stay RUNNING, refreshed per batch, until the captured //
|
||||
// deletes are applied; if a recovery takes the run over anyway, the drain //
|
||||
// stops, and it never reads rows captured after it started, which belong //
|
||||
// to a newer run //
|
||||
/////////////////////////////////////////////////////////////////////////////
|
||||
applyCapturedDeletes(client, tables, runs.values());
|
||||
|
||||
for(QuickSearchableTableConfig tableConfig : tables)
|
||||
{
|
||||
IndexCounts tableCounts = counts.get(tableConfig.getTableName());
|
||||
completeRunRecord(runs.get(tableConfig), RUN_COMPLETED, tableCounts.processed(), tableCounts.indexed(), 0, null);
|
||||
}
|
||||
|
||||
LOG.info("Full reindex complete", logPair("physicalIndex", physicalIndex), logPair("recordsIndexed", indexed));
|
||||
return (indexed);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
boolean takenOver = marked && isTakenOverOrUnknown(runs.values());
|
||||
LOG.warn(swapped ? "Full reindex failed after the alias swap; the new index stays in service" : "Full reindex failed; the previous index stays in service", e, logPair("physicalIndex", physicalIndex));
|
||||
|
||||
for(Map.Entry<QuickSearchableTableConfig, QuickSearchIndexRun> entry : runs.entrySet())
|
||||
{
|
||||
IndexCounts tableCounts = counts.get(entry.getKey().getTableName());
|
||||
completeRunRecord(entry.getValue(), RUN_FAILED,
|
||||
tableCounts == null ? 0 : tableCounts.processed(),
|
||||
tableCounts == null ? 0 : tableCounts.indexed(),
|
||||
tableCounts == null ? 1 : Math.max(1, tableCounts.errors()),
|
||||
e.getMessage());
|
||||
}
|
||||
|
||||
if(!swapped)
|
||||
if(created && !swapped)
|
||||
{
|
||||
try
|
||||
{
|
||||
@@ -223,8 +257,29 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
}
|
||||
}
|
||||
|
||||
restoreStatus(priorStatus);
|
||||
applyCapturedDeletes(client, tables);
|
||||
///////////////////////////////////////////////////////////////////////////
|
||||
// a recovery that took this run over owns its tables, captured deletes //
|
||||
// and run records (a newer run may own the first two by now); when that //
|
||||
// cannot be read, the run records stay RUNNING for the next recovery //
|
||||
///////////////////////////////////////////////////////////////////////////
|
||||
if(!takenOver)
|
||||
{
|
||||
if(marked)
|
||||
{
|
||||
restoreStatus(priorStatus);
|
||||
applyCapturedDeletes(client, tables, runs.values());
|
||||
}
|
||||
|
||||
for(Map.Entry<QuickSearchableTableConfig, QuickSearchIndexRun> entry : runs.entrySet())
|
||||
{
|
||||
IndexCounts tableCounts = counts.get(entry.getKey().getTableName());
|
||||
completeRunRecord(entry.getValue(), RUN_FAILED,
|
||||
tableCounts == null ? 0 : tableCounts.processed(),
|
||||
tableCounts == null ? 0 : tableCounts.indexed(),
|
||||
tableCounts == null ? 1 : Math.max(1, tableCounts.errors()),
|
||||
e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
if(e instanceof QException qException)
|
||||
{
|
||||
@@ -275,20 +330,421 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Recover what a full reindex that stopped without finishing (killed JVM,
|
||||
** lost node) left behind, unless a full reindex is alive. Returns the live
|
||||
** RUNNING FULL_REINDEX run records, and changes nothing when there are any.
|
||||
**
|
||||
** A run is alive while the startTime or modifyDate (its heartbeat) of one of
|
||||
** its RUNNING records is within fullReindexStaleMinutes. When none is
|
||||
** alive: each REBUILDING table goes back to ACTIVE (the status it had
|
||||
** before is not stored; drift detection marks NEEDS_REINDEX again where it
|
||||
** applies), with its lastBasepullTime moved back to the stopped run's
|
||||
** start, so basepull re-reads updates that a swap may have left out; with
|
||||
** handCapturedDeletesToBasepull, every AWAITING_REINDEX row up to the
|
||||
** newest one read here becomes PENDING; and only then is each stale RUNNING
|
||||
** record marked FAILED, so a recovery that fails partway leaves them for
|
||||
** the next one, run start included. Unlike completeRunRecord, that write
|
||||
** fails the recovery (and so the full reindex calling it) when any record
|
||||
** is not updated. Replaying those deletes is harmless
|
||||
** whether or not the stopped run swapped: before a swap the alias points at
|
||||
** the old index, which the listener already deleted from (a missing
|
||||
** document counts as success); after one, it removes the copy the run read
|
||||
** before the delete.
|
||||
**
|
||||
** The REBUILDING rows and captured rows are read before the run records. A
|
||||
** run creates its records before it marks a table, and the listener
|
||||
** captures a delete only once it sees REBUILDING, so anything read here
|
||||
** that belongs to a live run comes with a live record, and nothing of a
|
||||
** live run is touched. A run that starts after the records are read, or a
|
||||
** live run misjudged as stopped, can have a table put back to ACTIVE or its
|
||||
** records marked FAILED; it checks both right after a heartbeat, before its
|
||||
** swap, and fails without swapping. A misjudged run that already swapped
|
||||
** stops applying its captured deletes once its records are FAILED, and
|
||||
** never reads rows captured after its drain started.
|
||||
*******************************************************************************/
|
||||
List<QRecord> recoverDeadRun(boolean handCapturedDeletesToBasepull) throws QException
|
||||
{
|
||||
List<QRecord> rebuildingRows = queryRebuildingIndexRows();
|
||||
Serializable lastCapturedId = handCapturedDeletesToBasepull ? queryLastCapturedDeleteId() : null;
|
||||
List<QRecord> runningRuns = queryRunningRuns();
|
||||
|
||||
Instant cutoff = staleCutoff();
|
||||
List<QRecord> alive = runningRuns.stream().filter(run -> isAlive(run, cutoff)).toList();
|
||||
if(!alive.isEmpty() || (rebuildingRows.isEmpty() && runningRuns.isEmpty() && lastCapturedId == null))
|
||||
{
|
||||
return (alive);
|
||||
}
|
||||
|
||||
LOG.warn("Recovering what a full reindex that stopped without finishing left behind",
|
||||
logPair("staleRunIds", runningRuns.stream().map(run -> run.getValue("id")).toList()),
|
||||
logPair("tableNames", rebuildingRows.stream().map(row -> row.getValueString("tableName")).toList()));
|
||||
|
||||
for(QRecord row : rebuildingRows)
|
||||
{
|
||||
Map<String, Serializable> values = new HashMap<>();
|
||||
values.put("status", STATUS_ACTIVE);
|
||||
|
||||
Instant runStart = earliestStart(runningRuns, row.getValueInteger("id"));
|
||||
Instant lastBasepullTime = row.getValueInstant("lastBasepullTime");
|
||||
if(runStart != null && lastBasepullTime != null && runStart.isBefore(lastBasepullTime))
|
||||
{
|
||||
values.put("lastBasepullTime", runStart);
|
||||
}
|
||||
updateIndexRow(row.getValueInteger("id"), values);
|
||||
}
|
||||
|
||||
if(lastCapturedId != null)
|
||||
{
|
||||
handCapturedDeletesToBasepull(lastCapturedId);
|
||||
}
|
||||
|
||||
failDeadRuns(runningRuns);
|
||||
return (alive);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Mark the stale run records FAILED, failing when any of them is not
|
||||
** updated: a live run misjudged as stopped learns it was taken over only
|
||||
** from these records, so recovery must not finish, and no new full reindex
|
||||
** may start, until they are written. The status is mirrored onto each
|
||||
** index row, as completeRunRecord does.
|
||||
*******************************************************************************/
|
||||
private void failDeadRuns(List<QRecord> runningRuns) throws QException
|
||||
{
|
||||
if(runningRuns.isEmpty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Instant endTime = Instant.now();
|
||||
List<QRecord> updates = new ArrayList<>();
|
||||
for(QRecord runRecord : runningRuns)
|
||||
{
|
||||
updates.add(new QRecord()
|
||||
.withValue("id", runRecord.getValue("id"))
|
||||
.withValue("status", RUN_FAILED)
|
||||
.withValue("endTime", endTime)
|
||||
.withValue("recordsProcessed", 0)
|
||||
.withValue("recordsIndexed", 0)
|
||||
.withValue("errorCount", 1)
|
||||
.withValue("errorMessage", truncate(staleRunMessage(runRecord), 4000)));
|
||||
}
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName());
|
||||
updateInput.setRecords(updates);
|
||||
List<QRecord> results = CollectionUtils.nonNullList(new UpdateAction().execute(updateInput).getRecords());
|
||||
List<QRecord> written = results.stream().filter(result -> CollectionUtils.nullSafeIsEmpty(result.getErrors())).toList();
|
||||
if(written.size() < updates.size())
|
||||
{
|
||||
List<QRecord> notWritten = results.stream().filter(result -> CollectionUtils.nullSafeHasContents(result.getErrors())).toList();
|
||||
throw (new QException("Could not mark " + (updates.size() - written.size()) + " stopped full reindex run record(s) FAILED"
|
||||
+ (notWritten.isEmpty() ? "" : " (first error: " + notWritten.get(0).getErrors() + ")")
|
||||
+ "; recovery did not finish, and no full reindex starts until it does"));
|
||||
}
|
||||
|
||||
for(QRecord runRecord : runningRuns)
|
||||
{
|
||||
Integer indexId = runRecord.getValueInteger("quickSearchIndexId");
|
||||
if(indexId != null)
|
||||
{
|
||||
Map<String, Serializable> values = new HashMap<>();
|
||||
values.put("lastRunStatus", RUN_TYPE + " " + RUN_FAILED);
|
||||
values.put("lastErrorMessage", truncate(staleRunMessage(runRecord), 4000));
|
||||
updateIndexRow(indexId, values);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The error message recorded on a run that stopped without finishing.
|
||||
*******************************************************************************/
|
||||
private static String staleRunMessage(QRecord runRecord)
|
||||
{
|
||||
return ("Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Make the AWAITING_REINDEX rows up to lastCapturedId PENDING, for basepull
|
||||
** to replay. Rows are read in id order past the previous batch, so a row
|
||||
** that could not be updated is not read again.
|
||||
*******************************************************************************/
|
||||
private void handCapturedDeletesToBasepull(Serializable lastCapturedId) throws QException
|
||||
{
|
||||
QuickSearchQBitConfig config = getConfig();
|
||||
String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME);
|
||||
Serializable afterId = null;
|
||||
Integer handedOver = 0;
|
||||
|
||||
while(true)
|
||||
{
|
||||
QQueryFilter filter = new QQueryFilter(
|
||||
new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX),
|
||||
new QFilterCriteria("id", QCriteriaOperator.LESS_THAN_OR_EQUALS, lastCapturedId))
|
||||
.withOrderBy(new QFilterOrderBy("id", true))
|
||||
.withLimit(config.getSourceBatchSize());
|
||||
if(afterId != null)
|
||||
{
|
||||
filter.addCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, afterId));
|
||||
}
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(failedEventTable);
|
||||
queryInput.setFilter(filter);
|
||||
List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
|
||||
if(CollectionUtils.nullSafeIsEmpty(rows))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
afterId = rows.get(rows.size() - 1).getValue("id");
|
||||
List<QRecord> updates = new ArrayList<>();
|
||||
rows.forEach(row -> updates.add(new QRecord().withValue("id", row.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_PENDING)));
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(failedEventTable);
|
||||
updateInput.setRecords(updates);
|
||||
new UpdateAction().execute(updateInput);
|
||||
handedOver = handedOver + rows.size();
|
||||
}
|
||||
|
||||
LOG.info("Handed deletes captured during a stopped full reindex to basepull", logPair("count", handedOver));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fail when another full reindex is alive.
|
||||
*******************************************************************************/
|
||||
private void refuseIfAlive(List<QRecord> alive) throws QException
|
||||
{
|
||||
if(alive.isEmpty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
QRecord run = alive.get(0);
|
||||
throw (new QException("A full reindex is already running (run " + run.getValue("id") + ", last heartbeat " + lastSeen(run) + "); try again once it finishes. "
|
||||
+ "A run with no heartbeat for " + getConfig().getEffectiveFullReindexStaleMinutes() + " minutes is treated as stopped and recovered automatically."));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Just before the swap, fail unless this run still owns its tables: its
|
||||
** run records are still RUNNING (a recovery that judged it stopped marks
|
||||
** them FAILED) and every table is still REBUILDING (otherwise a delete
|
||||
** since then may have reached only the old index, uncaptured).
|
||||
*******************************************************************************/
|
||||
private void verifyStillOwned(List<QuickSearchableTableConfig> tables, Collection<QuickSearchIndexRun> runs) throws QException
|
||||
{
|
||||
if(isTakenOver(runs))
|
||||
{
|
||||
throw (new QException("Full reindex was recovered as stopped (no heartbeat for " + getConfig().getEffectiveFullReindexStaleMinutes() + " minutes) while still running; the alias was not swapped"));
|
||||
}
|
||||
|
||||
for(QuickSearchableTableConfig tableConfig : tables)
|
||||
{
|
||||
String status = queryIndexRow(tableConfig.getTableName()).getValueString("status");
|
||||
if(!STATUS_REBUILDING.equals(status))
|
||||
{
|
||||
throw (new QException("Table [" + tableConfig.getTableName() + "] is no longer REBUILDING (now " + status + "), so a delete since then may not have been captured; the alias was not swapped"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Refresh one of the run's records, which keeps the whole run alive.
|
||||
*******************************************************************************/
|
||||
private void keepAlive(Collection<QuickSearchIndexRun> runs)
|
||||
{
|
||||
runs.stream().findFirst().ifPresent(this::heartbeat);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Whether a recovery has marked any of the run records finished.
|
||||
*******************************************************************************/
|
||||
private boolean isTakenOver(Collection<QuickSearchIndexRun> runs) throws QException
|
||||
{
|
||||
if(runs.isEmpty())
|
||||
{
|
||||
return (false);
|
||||
}
|
||||
|
||||
List<Serializable> ids = new ArrayList<>();
|
||||
runs.forEach(run -> ids.add(run.getId()));
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(getConfig().getQuickSearchIndexRunTableName());
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria("id", QCriteriaOperator.IN, ids)));
|
||||
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()).stream()
|
||||
.anyMatch(run -> !RUN_RUNNING.equals(run.getValueString("status"))));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** isTakenOver for the failure path: never throws, and answers true when it
|
||||
** cannot tell, which leaves the tables to the next recovery.
|
||||
*******************************************************************************/
|
||||
private boolean isTakenOverOrUnknown(Collection<QuickSearchIndexRun> runs)
|
||||
{
|
||||
try
|
||||
{
|
||||
return (isTakenOver(runs));
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
LOG.warn("Could not read whether a recovery took over the failed full reindex; leaving its tables to the next recovery", e);
|
||||
return (true);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The live RUNNING FULL_REINDEX run records other than the given ones.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> queryLiveRunsOtherThan(Collection<QuickSearchIndexRun> own) throws QException
|
||||
{
|
||||
Set<Integer> ownIds = new HashSet<>();
|
||||
own.forEach(run -> ownIds.add(run.getId()));
|
||||
|
||||
Instant cutoff = staleCutoff();
|
||||
return (queryRunningRuns().stream()
|
||||
.filter(run -> !ownIds.contains(run.getValueInteger("id")) && isAlive(run, cutoff))
|
||||
.toList());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every RUNNING FULL_REINDEX run record, oldest first.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> queryRunningRuns() throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(getConfig().getQuickSearchIndexRunTableName());
|
||||
queryInput.setFilter(new QQueryFilter(
|
||||
new QFilterCriteria("runType", QCriteriaOperator.EQUALS, RUN_TYPE),
|
||||
new QFilterCriteria("status", QCriteriaOperator.EQUALS, RUN_RUNNING))
|
||||
.withOrderBy(new QFilterOrderBy("id", true)));
|
||||
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every quickSearchIndex row that is REBUILDING.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> queryRebuildingIndexRows() throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(getConfig().getQuickSearchIndexTableName());
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, STATUS_REBUILDING)));
|
||||
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The id of the newest AWAITING_REINDEX row, or null. Ids grow in insert
|
||||
** order, so a row captured after this read is beyond it.
|
||||
*******************************************************************************/
|
||||
private Serializable queryLastCapturedDeleteId() throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME));
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX))
|
||||
.withOrderBy(new QFilterOrderBy("id", false))
|
||||
.withLimit(1));
|
||||
List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
|
||||
return (CollectionUtils.nullSafeIsEmpty(rows) ? null : rows.get(0).getValue("id"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Run records last seen before this are stale.
|
||||
*******************************************************************************/
|
||||
private Instant staleCutoff() throws QException
|
||||
{
|
||||
return (Instant.now().minus(getConfig().getEffectiveFullReindexStaleMinutes(), ChronoUnit.MINUTES));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Whether a run record was last seen after the cutoff.
|
||||
*******************************************************************************/
|
||||
private static boolean isAlive(QRecord run, Instant cutoff)
|
||||
{
|
||||
Instant lastSeen = lastSeen(run);
|
||||
return (lastSeen != null && lastSeen.isAfter(cutoff));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A run record's latest sign of life: its startTime or its heartbeat
|
||||
** (modifyDate), whichever is later; null when it has neither.
|
||||
*******************************************************************************/
|
||||
private static Instant lastSeen(QRecord run)
|
||||
{
|
||||
Instant startTime = run.getValueInstant("startTime");
|
||||
Instant modifyDate = run.getValueInstant("modifyDate");
|
||||
if(startTime == null || modifyDate == null)
|
||||
{
|
||||
return (startTime == null ? modifyDate : startTime);
|
||||
}
|
||||
return (startTime.isAfter(modifyDate) ? startTime : modifyDate);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The earliest startTime among the run records of an index row, or null.
|
||||
*******************************************************************************/
|
||||
private static Instant earliestStart(List<QRecord> runs, Integer indexId)
|
||||
{
|
||||
return (runs.stream()
|
||||
.filter(run -> Objects.equals(indexId, run.getValueInteger("quickSearchIndexId")))
|
||||
.map(run -> run.getValueInstant("startTime"))
|
||||
.filter(Objects::nonNull)
|
||||
.min(Instant::compareTo)
|
||||
.orElse(null));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Apply the deletes the listener captured during the rebuild (failed-event
|
||||
** rows with status AWAITING_REINDEX) through the alias, after the rebuild's
|
||||
** own writes, so an in-flight page cannot bring a deleted record back.
|
||||
** Called once the tables are no longer REBUILDING, so the listener stops
|
||||
** adding rows (one that races the status change hands its own rows to
|
||||
** basepull).
|
||||
** basepull). Only rows up to the newest one read at the start are applied,
|
||||
** since a later row belongs to a newer run. The run is kept alive before
|
||||
** every batch, and the drain stops once a recovery has taken it over.
|
||||
**
|
||||
** Rows are read in id order, past the previous batch, so a batch whose rows
|
||||
** could not be removed or updated (QQQ reports those as per-row errors, not
|
||||
** exceptions) is not read again; such rows stay AWAITING_REINDEX for the next
|
||||
** full reindex. Never throws.
|
||||
** exceptions) is not read again; such rows stay AWAITING_REINDEX until the
|
||||
** next full reindex, or a basepull that finds none running, picks them up.
|
||||
** Never throws.
|
||||
*******************************************************************************/
|
||||
void applyCapturedDeletes(QuickSearchOpenSearchClient client, List<QuickSearchableTableConfig> tables)
|
||||
void applyCapturedDeletes(QuickSearchOpenSearchClient client, List<QuickSearchableTableConfig> tables, Collection<QuickSearchIndexRun> runs)
|
||||
{
|
||||
List<Serializable> tableNames = new ArrayList<>();
|
||||
tables.forEach(tableConfig -> tableNames.add(tableConfig.getTableName()));
|
||||
@@ -302,13 +758,22 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
{
|
||||
QuickSearchQBitConfig config = getConfig();
|
||||
String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME);
|
||||
Serializable lastCapturedId = queryLastCapturedDeleteId();
|
||||
Serializable afterId = null;
|
||||
|
||||
while(true)
|
||||
while(lastCapturedId != null)
|
||||
{
|
||||
if(isTakenOver(runs))
|
||||
{
|
||||
LOG.warn("A recovery took this full reindex over while it applied its captured deletes; leaving the rest to the recovery", logPair("applied", applied));
|
||||
break;
|
||||
}
|
||||
keepAlive(runs);
|
||||
|
||||
QQueryFilter filter = new QQueryFilter(
|
||||
new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX),
|
||||
new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames))
|
||||
new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames),
|
||||
new QFilterCriteria("id", QCriteriaOperator.LESS_THAN_OR_EQUALS, lastCapturedId))
|
||||
.withOrderBy(new QFilterOrderBy("id", true))
|
||||
.withLimit(config.getSourceBatchSize());
|
||||
if(afterId != null)
|
||||
@@ -449,35 +914,6 @@ public class FullReindexStep extends AbstractIndexingStep
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The record IDs, among the rows' recordIds, that exist in the source table.
|
||||
*******************************************************************************/
|
||||
Set<String> queryExistingRecordIds(String tableName, List<QRecord> rows) throws QException
|
||||
{
|
||||
QuickSearchableTableConfig tableConfig = getTableConfig(tableName);
|
||||
if(tableConfig == null)
|
||||
{
|
||||
return (Set.of());
|
||||
}
|
||||
|
||||
String primaryKeyField = tableConfig.getPrimaryKeyField() == null ? "id" : tableConfig.getPrimaryKeyField();
|
||||
List<Serializable> recordIds = new ArrayList<>();
|
||||
rows.forEach(row -> recordIds.add(row.getValueString("recordId")));
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(tableName);
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, recordIds)));
|
||||
|
||||
Set<String> existingIds = new HashSet<>();
|
||||
for(QRecord record : CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()))
|
||||
{
|
||||
existingIds.add(record.getValueString(primaryKeyField));
|
||||
}
|
||||
return (existingIds);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Rebuild one table in place (kept for callers that used this directly).
|
||||
*******************************************************************************/
|
||||
|
||||
@@ -132,7 +132,7 @@ public class ReconcileIndexStep extends AbstractIndexingStep
|
||||
try
|
||||
{
|
||||
Instant reconcileStart = Instant.now();
|
||||
IndexCounts counts = indexAllRecords(client, tableConfig, List.of(), null);
|
||||
IndexCounts counts = indexAllRecords(client, tableConfig, List.of(), null, run);
|
||||
|
||||
Long documentsRemoved = 0L;
|
||||
if(counts.errors() == 0)
|
||||
|
||||
+3
-2
@@ -47,6 +47,7 @@ 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.model.QuickSearchFailedEvent;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
@@ -176,9 +177,9 @@ class FullReindexDeleteReplayIntegrationTest
|
||||
// while the alias still points at the old one //
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
@Override
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex) throws QException
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex);
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run);
|
||||
|
||||
DeleteInput deleteInput = new DeleteInput();
|
||||
deleteInput.setTableName(TABLE_NAME);
|
||||
|
||||
+321
@@ -0,0 +1,321 @@
|
||||
/*
|
||||
* 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.QueryAction;
|
||||
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.query.QFilterCriteria;
|
||||
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.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.model.QuickSearchFailedEvent;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep;
|
||||
import com.kingsrook.qbits.quicksearch.processes.BasepullIndexStep;
|
||||
import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
|
||||
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;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Integration test, against a real OpenSearch, that a full reindex killed
|
||||
** right after its alias swap (table left REBUILDING, RUNNING run record, a
|
||||
** captured delete never applied, so the deleted record is back in search) is
|
||||
** recovered by the scheduled basepull once its run record goes stale.
|
||||
*******************************************************************************/
|
||||
@ExtendWith(RequiresDockerCondition.class)
|
||||
@Testcontainers
|
||||
class FullReindexRecoveryIntegrationTest
|
||||
{
|
||||
private static final String BACKEND_NAME = "reindexRecoveryMemoryBackend";
|
||||
private static final String TABLE_NAME = "recoveryCustomer";
|
||||
private static final String INDEX_NAME = "quick_search_reindex_recovery_test";
|
||||
|
||||
@Container
|
||||
static GenericContainer<?> opensearch = new GenericContainer<>(System.getProperty("opensearch.test.image", "opensearchproject/opensearch:2.19.6"))
|
||||
.withExposedPorts(9200)
|
||||
.withEnv("discovery.type", "single-node")
|
||||
.withEnv("DISABLE_SECURITY_PLUGIN", "true")
|
||||
.withEnv("DISABLE_INSTALL_DEMO_CONFIG", "true")
|
||||
.waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200));
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Entity annotated for quick search indexing.
|
||||
*******************************************************************************/
|
||||
@QuickSearchable(tableName = TABLE_NAME)
|
||||
static class RecoveryCustomer
|
||||
{
|
||||
@QuickSearchField(weight = 2)
|
||||
private String name;
|
||||
|
||||
@QuickSearchField
|
||||
private String city;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Thrown to stop a full reindex the way a killed JVM does: it is not an
|
||||
** Exception, so none of the step's failure handling runs.
|
||||
*******************************************************************************/
|
||||
static class SimulatedKill extends Error
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Build the instance, produce the QBit with real-time indexing on and the
|
||||
** record-security re-read off (search returns whatever the index holds),
|
||||
** 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))
|
||||
.withField(new QFieldMetaData("modifyDate", QFieldType.DATE_TIME)));
|
||||
|
||||
QuickSearchQBitConfig config = new QuickSearchQBitConfig()
|
||||
.withBackendName(BACKEND_NAME)
|
||||
.withOpensearchHost(opensearch.getHost())
|
||||
.withOpensearchPort(opensearch.getMappedPort(9200))
|
||||
.withOpensearchIndexName(INDEX_NAME)
|
||||
.withSearchableEntityClasses(List.of(RecoveryCustomer.class))
|
||||
.withEnableRealTimeIndexing(true)
|
||||
.withApplyRecordSecurityLocks(false);
|
||||
|
||||
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 record deleted during a full reindex that is then killed right
|
||||
** after its swap is back in search (the new index holds the copy read
|
||||
** before the delete, and the captured delete was never applied). While the
|
||||
** dead run's record is fresh it blocks a new full reindex; once it is
|
||||
** stale, basepull recovers the table and replays the delete, and the
|
||||
** record is gone from search.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testKilledFullReindex_isRecoveredByBasepull() throws Exception
|
||||
{
|
||||
InsertInput insertInput = new InsertInput();
|
||||
insertInput.setTableName(TABLE_NAME);
|
||||
insertInput.setRecords(List.of(
|
||||
new QRecord().withValue("id", 1).withValue("name", "Gwendolyn").withValue("city", "Stalledton"),
|
||||
new QRecord().withValue("id", 2).withValue("name", "Stanford").withValue("city", "Stalledton")));
|
||||
new InsertAction().execute(insertInput);
|
||||
assertThat(searchRecordIds("stalledton")).containsExactlyInAnyOrder("1", "2");
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
////////////////////////////////////////////////////////////////////
|
||||
// delete a record once its document is in the new physical index //
|
||||
////////////////////////////////////////////////////////////////////
|
||||
@Override
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run);
|
||||
|
||||
DeleteInput deleteInput = new DeleteInput();
|
||||
deleteInput.setTableName(TABLE_NAME);
|
||||
deleteInput.setPrimaryKeys(List.<Serializable>of(1));
|
||||
new DeleteAction().execute(deleteInput);
|
||||
|
||||
return (counts);
|
||||
}
|
||||
|
||||
|
||||
|
||||
////////////////////////////////////////////////////////////////////
|
||||
// die right after the swap, before any table is finished //
|
||||
////////////////////////////////////////////////////////////////////
|
||||
@Override
|
||||
protected void updateIndexRow(Integer indexId, Map<String, ? extends Serializable> values) throws QException
|
||||
{
|
||||
if(values.containsKey("lastFullReindexTime"))
|
||||
{
|
||||
throw (new SimulatedKill());
|
||||
}
|
||||
super.updateIndexRow(indexId, values);
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(SimulatedKill.class);
|
||||
|
||||
assertThat(indexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement()
|
||||
.satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING));
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement()
|
||||
.satisfies(row -> assertThat(row.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX));
|
||||
assertThat(searchRecordIds("stalledton"))
|
||||
.as("the swapped-in index still holds the deleted record")
|
||||
.containsExactlyInAnyOrder("1", "2");
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.as("the dead run's record is still fresh, so it counts as running")
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("already running");
|
||||
|
||||
backdateRunRecords();
|
||||
new BasepullIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
assertThat(searchRecordIds("stalledton")).containsExactly("2");
|
||||
assertThat(indexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME))
|
||||
.filteredOn(run -> FullReindexStep.RUN_TYPE.equals(run.getValueString("runType")))
|
||||
.singleElement()
|
||||
.satisfies(run ->
|
||||
{
|
||||
assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED);
|
||||
assertThat(run.getValueString("errorMessage")).contains("stopped without finishing");
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Move the start and heartbeat of every RUNNING run record two hours into
|
||||
** the past, as if its run had stopped refreshing it.
|
||||
*******************************************************************************/
|
||||
private void backdateRunRecords() throws QException
|
||||
{
|
||||
Instant twoHoursAgo = Instant.now().minus(2, ChronoUnit.HOURS);
|
||||
for(QRecord run : queryAll(QuickSearchIndexRun.TABLE_NAME))
|
||||
{
|
||||
if(AbstractIndexingStep.RUN_RUNNING.equals(run.getValueString("status")))
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME);
|
||||
updateInput.setOmitModifyDateUpdate(true);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", run.getValue("id")).withValue("startTime", twoHoursAgo).withValue("modifyDate", twoHoursAgo)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The table's quickSearchIndex row.
|
||||
*******************************************************************************/
|
||||
private QRecord indexRow() throws QException
|
||||
{
|
||||
return (queryAll(QuickSearchIndex.TABLE_NAME).stream()
|
||||
.filter(row -> TABLE_NAME.equals(row.getValueString("tableName")))
|
||||
.findFirst().orElseThrow());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** All rows of a table.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> queryAll(String tableName) throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(tableName);
|
||||
return (new QueryAction().execute(queryInput).getRecords());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Refresh the index, then return the record IDs search returns for the term.
|
||||
*******************************************************************************/
|
||||
private List<String> searchRecordIds(String term) throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getClient().refreshIndex();
|
||||
return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME))
|
||||
.getResults().stream()
|
||||
.map(QuickSearchResult::getRecordId)
|
||||
.toList());
|
||||
}
|
||||
|
||||
}
|
||||
+14
@@ -347,6 +347,7 @@ class QuickSearchQBitConfigValidationTest
|
||||
.withBasepullRepeatSeconds(0)
|
||||
.withBasepullOverlapSeconds(-5)
|
||||
.withRunHistoryRetentionDays(0)
|
||||
.withFullReindexStaleMinutes(0)
|
||||
.withMaxFieldLength(0));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("connectTimeoutMillis"));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("responseTimeoutMillis"));
|
||||
@@ -356,11 +357,23 @@ class QuickSearchQBitConfigValidationTest
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("basepullRepeatSeconds"));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("basepullOverlapSeconds"));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("runHistoryRetentionDays"));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("fullReindexStaleMinutes"));
|
||||
assertThat(errors).anyMatch(e -> e.startsWith("maxFieldLength"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Test
|
||||
void testFullReindexStaleMinutes_nullIsValid_andMeansTheDefault()
|
||||
{
|
||||
QuickSearchQBitConfig config = baseConfig().withFullReindexStaleMinutes(null);
|
||||
assertThat(validate(config)).isEmpty();
|
||||
assertThat(config.getEffectiveFullReindexStaleMinutes()).isEqualTo(QuickSearchQBitConfig.DEFAULT_FULL_REINDEX_STALE_MINUTES);
|
||||
assertThat(config.withFullReindexStaleMinutes(5).getEffectiveFullReindexStaleMinutes()).isEqualTo(5);
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Test
|
||||
void testValidate_nullInstance_isSafe()
|
||||
{
|
||||
@@ -406,6 +419,7 @@ class QuickSearchQBitConfigValidationTest
|
||||
assertThat(config.getResponseTimeoutMillis()).isEqualTo(60000);
|
||||
assertThat(config.getMaxSearchLimit()).isEqualTo(100);
|
||||
assertThat(config.getBasepullOverlapSeconds()).isEqualTo(300);
|
||||
assertThat(config.getFullReindexStaleMinutes()).isEqualTo(30);
|
||||
assertThat(config.getAdminPermissionRules().getPermissionBaseName()).isEqualTo(QuickSearchQBitConfig.DEFAULT_ADMIN_PERMISSION_BASE_NAME);
|
||||
assertThat(config.getApplyRecordSecurityLocks()).isTrue();
|
||||
assertThat(config.getDefaultBackendNameForTables()).isNull();
|
||||
|
||||
@@ -224,6 +224,34 @@ class BasepullIndexStepTest extends BaseQuickSearchTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Insert a RUNNING FULL_REINDEX run record for the index row whose start
|
||||
** and last heartbeat were the given number of minutes ago.
|
||||
*******************************************************************************/
|
||||
private QRecord insertRunningFullReindex(QRecord indexRow, long minutesAgo) throws QException
|
||||
{
|
||||
InsertInput input = new InsertInput();
|
||||
input.setTableName(QuickSearchIndexRun.TABLE_NAME);
|
||||
input.setRecords(List.of(new QRecord()
|
||||
.withValue("quickSearchIndexId", indexRow.getValue("id"))
|
||||
.withValue("runType", FullReindexStep.RUN_TYPE)
|
||||
.withValue("status", AbstractIndexingStep.RUN_RUNNING)));
|
||||
QRecord run = new InsertAction().execute(input).getRecords().get(0);
|
||||
|
||||
////////////////////////////////////////////////////////////////////
|
||||
// set both times without QQQ stamping modifyDate with the present //
|
||||
////////////////////////////////////////////////////////////////////
|
||||
Instant lastSeen = Instant.now().minus(minutesAgo, ChronoUnit.MINUTES);
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME);
|
||||
updateInput.setOmitModifyDateUpdate(true);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", run.getValue("id")).withValue("startTime", lastSeen).withValue("modifyDate", lastSeen)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
return (run);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query all rows of a table.
|
||||
*******************************************************************************/
|
||||
@@ -923,7 +951,8 @@ class BasepullIndexStepTest extends BaseQuickSearchTest
|
||||
@Test
|
||||
void testReplayFailedEvents_awaitingReindexRowsAreNotReplayed() throws QException
|
||||
{
|
||||
insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
|
||||
QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
|
||||
insertRunningFullReindex(row, 1);
|
||||
QRecord captured = insertFailedEvent(TEST_ENTITY_TABLE, "7", "DELETE", 0);
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
|
||||
@@ -940,6 +969,65 @@ class BasepullIndexStepTest extends BaseQuickSearchTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a DELETE event whose record exists again (re-created with the same
|
||||
** key since) keeps the record's document; the event is resolved.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReplayFailedEvents_deleteOfRecreatedRecord_keepsItsDocument() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
|
||||
insertFailedEvent(TEST_ENTITY_TABLE, "1", "DELETE", 0);
|
||||
insertFailedEvent(TEST_ENTITY_TABLE, "9", "DELETE", 0);
|
||||
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new BasepullIndexStep().run(new RunBackendStepInput(), output);
|
||||
|
||||
verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt());
|
||||
verify(mockClient).deleteDocuments(List.of(TEST_ENTITY_TABLE + ":9"), QuickSearchQBitContext.getConfig().getBulkBatchSize());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
|
||||
assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(2);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a full reindex that stopped without finishing (table left
|
||||
** REBUILDING, run record without a heartbeat for two hours) is recovered
|
||||
** by the scheduled basepull, which replays its captured deletes in the same
|
||||
** run.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testBasepull_recoversAStoppedFullReindex_andReplaysItsCapturedDeletes() throws QException
|
||||
{
|
||||
QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
|
||||
QRecord deadRun = insertRunningFullReindex(row, 120);
|
||||
QRecord captured = insertFailedEvent(TEST_ENTITY_TABLE, "7", "DELETE", 0);
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", captured.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", AbstractIndexingStep.STATUS_REBUILDING)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new BasepullIndexStep().run(new RunBackendStepInput(), output);
|
||||
|
||||
verify(mockClient).deleteDocuments(List.of(TEST_ENTITY_TABLE + ":7"), QuickSearchQBitContext.getConfig().getBulkBatchSize());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
|
||||
assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(1);
|
||||
assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME))
|
||||
.filteredOn(run -> deadRun.getValue("id").equals(run.getValue("id")))
|
||||
.singleElement()
|
||||
.satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: drift found while a full reindex is running does not replace the
|
||||
** REBUILDING status the listener relies on.
|
||||
@@ -948,6 +1036,7 @@ class BasepullIndexStepTest extends BaseQuickSearchTest
|
||||
void testDriftDetection_duringFullReindex_keepsRebuilding() throws QException
|
||||
{
|
||||
QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
|
||||
insertRunningFullReindex(row, 1);
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id"))
|
||||
|
||||
@@ -17,9 +17,14 @@
|
||||
package com.kingsrook.qbits.quicksearch.processes;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
@@ -37,9 +42,11 @@ 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.actions.tables.update.UpdateOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.statusmessages.SystemErrorStatusMessage;
|
||||
import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
|
||||
@@ -61,6 +68,7 @@ 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.doAnswer;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
@@ -556,9 +564,9 @@ class FullReindexStepTest extends BaseQuickSearchTest
|
||||
new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex) throws QException
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex);
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run);
|
||||
statusesDuringRebuild.add(FullReindexStepTest.this.queryIndexRow().getValueString("status"));
|
||||
insertCapturedDelete("7");
|
||||
return (counts);
|
||||
@@ -751,4 +759,777 @@ class FullReindexStepTest extends BaseQuickSearchTest
|
||||
assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Insert a RUNNING FULL_REINDEX run record for the given index row, started
|
||||
** startedMinutesAgo and last refreshed (heartbeat) heartbeatMinutesAgo.
|
||||
*******************************************************************************/
|
||||
private QRecord insertRunningRun(Integer quickSearchIndexId, long startedMinutesAgo, long heartbeatMinutesAgo) throws QException
|
||||
{
|
||||
InsertInput insertInput = new InsertInput();
|
||||
insertInput.setTableName(QuickSearchIndexRun.TABLE_NAME);
|
||||
insertInput.setRecords(List.of(new QRecord()
|
||||
.withValue("quickSearchIndexId", quickSearchIndexId)
|
||||
.withValue("runType", FullReindexStep.RUN_TYPE)
|
||||
.withValue("status", AbstractIndexingStep.RUN_RUNNING)));
|
||||
QRecord run = new InsertAction().execute(insertInput).getRecords().get(0);
|
||||
setRunTimes(run.getValueInteger("id"), Instant.now().minus(startedMinutesAgo, ChronoUnit.MINUTES), Instant.now().minus(heartbeatMinutesAgo, ChronoUnit.MINUTES));
|
||||
return (queryRun(run.getValueInteger("id")));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Move a run record's start and heartbeat two hours into the past, as if
|
||||
** its run had stopped refreshing it.
|
||||
*******************************************************************************/
|
||||
private void backdateRun(Integer runId) throws QException
|
||||
{
|
||||
Instant twoHoursAgo = Instant.now().minus(2, ChronoUnit.HOURS);
|
||||
setRunTimes(runId, twoHoursAgo, twoHoursAgo);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Set a run record's startTime and modifyDate, without QQQ stamping the
|
||||
** modifyDate with the current time.
|
||||
*******************************************************************************/
|
||||
private void setRunTimes(Integer runId, Instant startTime, Instant modifyDate) throws QException
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME);
|
||||
updateInput.setOmitModifyDateUpdate(true);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", runId).withValue("startTime", startTime).withValue("modifyDate", modifyDate)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The run record with the given id, for lambdas and overrides that cannot
|
||||
** throw a checked exception.
|
||||
*******************************************************************************/
|
||||
private QRecord queryRunQuietly(Integer runId)
|
||||
{
|
||||
try
|
||||
{
|
||||
return (queryRun(runId));
|
||||
}
|
||||
catch(QException e)
|
||||
{
|
||||
throw (new IllegalStateException(e));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The run record with the given id.
|
||||
*******************************************************************************/
|
||||
private QRecord queryRun(Integer runId) throws QException
|
||||
{
|
||||
return (queryAll(QuickSearchIndexRun.TABLE_NAME).stream()
|
||||
.filter(run -> runId.equals(run.getValueInteger("id")))
|
||||
.findFirst().orElseThrow());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Set the status and lastBasepullTime on the testEntity quickSearchIndex row.
|
||||
*******************************************************************************/
|
||||
private void updateIndexRow(String status, Instant lastBasepullTime) throws QException
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", queryIndexRow().getValue("id"))
|
||||
.withValue("status", status)
|
||||
.withValue("lastBasepullTime", lastBasepullTime)));
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a full reindex whose run record is fresh blocks a second one, which
|
||||
** fails before it creates anything.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testLiveRun_blocksASecondRun() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
insertIndexRow(true);
|
||||
insertRunningRun(queryIndexRow().getValueInteger("id"), 1, 1);
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("already running");
|
||||
|
||||
verify(mockClient, never()).createPhysicalIndex(anyString());
|
||||
verify(mockClient, never()).indexDocuments(anyString(), anyList(), anyInt());
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement()
|
||||
.satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a live full reindex also blocks a single-table full reindex.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testLiveRun_blocksASingleTableRun() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
insertIndexRow(true);
|
||||
insertRunningRun(queryIndexRow().getValueInteger("id"), 1, 1);
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep().run(inputForTable(TEST_ENTITY_TABLE), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("already running");
|
||||
|
||||
verify(mockClient, never()).indexDocuments(anyList(), anyInt());
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).hasSize(1);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a dead run (no heartbeat past the threshold) does not block a new
|
||||
** one; it is recovered first, and the deletes it captured wait for the new
|
||||
** run, which applies them after its swap.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testStaleRun_doesNotBlock_andIsRecovered() throws QException
|
||||
{
|
||||
insertTestEntities(2);
|
||||
insertIndexRow(true);
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
Integer staleRunId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id");
|
||||
insertCapturedDelete("9");
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
|
||||
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new FullReindexStep().run(new RunBackendStepInput(), output);
|
||||
|
||||
assertThat(output.getValueBoolean("aliasSwapped")).isTrue();
|
||||
assertThat(queryRun(staleRunId).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED);
|
||||
assertThat(queryRun(staleRunId).getValueString("errorMessage")).contains("stopped without finishing");
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
|
||||
InOrder inOrder = inOrder(mockClient);
|
||||
inOrder.verify(mockClient).swapAliasTo(PHYSICAL_INDEX);
|
||||
inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the run records stay RUNNING while the captured deletes are applied
|
||||
** after the swap, so a basepull meanwhile leaves those rows to this run.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testAllTables_runStaysRunningUntilCapturedDeletesAreApplied() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
insertCapturedDelete("9");
|
||||
|
||||
List<String> runStatusesWhileApplying = new ArrayList<>();
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
queryAll(QuickSearchIndexRun.TABLE_NAME).forEach(run -> runStatusesWhileApplying.add(run.getValueString("status")));
|
||||
return (new BulkIndexResult().withSuccessCount(1));
|
||||
});
|
||||
|
||||
new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
assertThat(runStatusesWhileApplying).containsExactly(AbstractIndexingStep.RUN_RUNNING);
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement()
|
||||
.satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_COMPLETED));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the run record's modifyDate is refreshed after every page, so a
|
||||
** long rebuild keeps showing it is alive.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testHeartbeat_refreshesTheRunRecordAfterEachPage() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(2);
|
||||
insertTestEntities(5);
|
||||
insertIndexRow(true);
|
||||
Integer runId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id");
|
||||
|
||||
AtomicInteger pagesIndexed = new AtomicInteger(0);
|
||||
when(mockClient.indexDocuments(anyString(), anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
pagesIndexed.incrementAndGet();
|
||||
return (new BulkIndexResult().withSuccessCount(((List<?>) invocation.getArgument(1)).size()));
|
||||
});
|
||||
|
||||
List<Integer> pagesIndexedAtEachHeartbeat = new ArrayList<>();
|
||||
List<Instant> heartbeatsWritten = new ArrayList<>();
|
||||
FullReindexStep step = new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected void heartbeat(QuickSearchIndexRun run)
|
||||
{
|
||||
super.heartbeat(run);
|
||||
pagesIndexedAtEachHeartbeat.add(pagesIndexed.get());
|
||||
heartbeatsWritten.add(queryRunQuietly(runId).getValueInstant("modifyDate"));
|
||||
}
|
||||
};
|
||||
|
||||
Instant before = Instant.now();
|
||||
step.indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX,
|
||||
new QuickSearchIndexRun().withId(runId).withRunType(FullReindexStep.RUN_TYPE));
|
||||
|
||||
assertThat(pagesIndexedAtEachHeartbeat).containsExactly(1, 2, 3);
|
||||
assertThat(heartbeatsWritten).hasSize(3).allSatisfy(heartbeat -> assertThat(heartbeat).isAfterOrEqualTo(before));
|
||||
assertThat(heartbeatsWritten).isSorted();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: only FULL_REINDEX runs have a heartbeat (nothing reads one for the
|
||||
** other run types).
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testHeartbeat_onlyForFullReindexRuns() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
QRecord run = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120);
|
||||
|
||||
new FullReindexStep().heartbeat(new QuickSearchIndexRun().withId(run.getValueInteger("id")).withRunType(BasepullIndexStep.RUN_TYPE));
|
||||
assertThat(queryRun(run.getValueInteger("id")).getValueInstant("modifyDate")).isEqualTo(run.getValueInstant("modifyDate"));
|
||||
|
||||
new FullReindexStep().heartbeat(new QuickSearchIndexRun().withId(run.getValueInteger("id")).withRunType(FullReindexStep.RUN_TYPE));
|
||||
assertThat(queryRun(run.getValueInteger("id")).getValueInstant("modifyDate")).isAfter(run.getValueInstant("modifyDate"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a heartbeat that cannot be written is logged and the pages are
|
||||
** still indexed.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testHeartbeat_failureDoesNotStopIndexing() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(2);
|
||||
insertTestEntities(3);
|
||||
|
||||
AbstractIndexingStep.IndexCounts counts;
|
||||
try(MockedConstruction<UpdateAction> updateActions = mockConstruction(UpdateAction.class, (updateAction, context) ->
|
||||
when(updateAction.execute(any())).thenThrow(new QException("database unavailable"))))
|
||||
{
|
||||
counts = new FullReindexStep().indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX,
|
||||
new QuickSearchIndexRun().withId(1).withRunType(FullReindexStep.RUN_TYPE));
|
||||
assertThat(updateActions.constructed()).hasSize(2);
|
||||
}
|
||||
|
||||
assertThat(counts.processed()).isEqualTo(3);
|
||||
assertThat(counts.indexed()).isEqualTo(3);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a run that stalls past the threshold is recovered by another node
|
||||
** while still going. It notices before its swap and stops without swapping
|
||||
** and without touching the tables or the captured deletes, which the
|
||||
** recovery already handed to basepull.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRunJudgedDead_recoveredWhileGoing_doesNotSwap() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run);
|
||||
insertCapturedDelete("7");
|
||||
backdateRun(run.getId());
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
return (counts);
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("recovered");
|
||||
|
||||
verify(mockClient, never()).swapAliasTo(anyString());
|
||||
verify(mockClient).deletePhysicalIndex(PHYSICAL_INDEX);
|
||||
verify(mockClient, never()).deleteDocuments(anyList(), anyInt());
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement()
|
||||
.satisfies(row -> assertThat(row.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a table that is no longer REBUILDING before the swap (a recovery
|
||||
** that read the run records just before this run created its own put it
|
||||
** back to ACTIVE) means a delete since may have reached only the old index,
|
||||
** so the run fails without swapping and applies its captured deletes to the
|
||||
** old index.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testTableNoLongerRebuildingBeforeTheSwap_doesNotSwap() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException
|
||||
{
|
||||
IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run);
|
||||
insertCapturedDelete("7");
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
return (counts);
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("no longer REBUILDING");
|
||||
|
||||
verify(mockClient, never()).swapAliasTo(anyString());
|
||||
verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":7")), anyInt());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement()
|
||||
.satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: when another node starts a full reindex at the same moment (after
|
||||
** this one checked, before it created its run records), this one sees the
|
||||
** other's fresh run record and stops before it touches anything.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testSimultaneousStart_secondRunStopsBeforeMarking() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
AtomicBoolean otherStarted = new AtomicBoolean(false);
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected QuickSearchIndexRun createRunRecord(Integer indexId, String runType) throws QException
|
||||
{
|
||||
if(otherStarted.compareAndSet(false, true))
|
||||
{
|
||||
insertRunningRun(indexId, 0, 0);
|
||||
}
|
||||
return (super.createRunRecord(indexId, runType));
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("already running");
|
||||
|
||||
verify(mockClient, never()).createPhysicalIndex(anyString());
|
||||
verify(mockClient, never()).swapAliasTo(anyString());
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).extracting(run -> run.getValueString("status"))
|
||||
.containsExactlyInAnyOrder(AbstractIndexingStep.RUN_RUNNING, AbstractIndexingStep.RUN_FAILED);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: recovering a dead run marks its record FAILED, puts the table back
|
||||
** to ACTIVE with the basepull watermark rewound to the run's start, and
|
||||
** hands the captured deletes to basepull.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_deadRun_restoresTableAndHandsCapturedDeletesToBasepull() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, Instant.now().minus(5, ChronoUnit.MINUTES));
|
||||
QRecord deadRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 90);
|
||||
insertCapturedDelete("4");
|
||||
insertCapturedDelete("5");
|
||||
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
|
||||
QRecord row = queryIndexRow();
|
||||
assertThat(row.getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(row.getValueInstant("lastBasepullTime")).isEqualTo(deadRun.getValueInstant("startTime"));
|
||||
assertThat(row.getValueString("lastRunStatus")).isEqualTo("FULL_REINDEX FAILED");
|
||||
assertThat(row.getValueString("lastErrorMessage")).contains("stopped without finishing");
|
||||
|
||||
QRecord run = queryRun(deadRun.getValueInteger("id"));
|
||||
assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED);
|
||||
assertThat(run.getValue("endTime")).isNotNull();
|
||||
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("status"))
|
||||
.containsExactly(QuickSearchFailedEvent.STATUS_PENDING, QuickSearchFailedEvent.STATUS_PENDING);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a run started long ago but with a recent heartbeat is alive;
|
||||
** recovery returns it and leaves its table, run record and captured
|
||||
** deletes alone.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_liveRun_leavesItsTableAndRowsAlone() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
QRecord liveRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 180, 1);
|
||||
insertCapturedDelete("4");
|
||||
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).extracting(run -> run.getValueInteger("id"))
|
||||
.containsExactly(liveRun.getValueInteger("id"));
|
||||
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
assertThat(queryRun(liveRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING);
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement()
|
||||
.satisfies(event -> assertThat(event.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a REBUILDING table whose run record is gone (purged) is recovered
|
||||
** too, without moving its basepull watermark.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_withoutRunRecord_restoresTheTable() throws QException
|
||||
{
|
||||
Instant lastBasepullTime = Instant.now().minus(5, ChronoUnit.MINUTES);
|
||||
insertIndexRow(true);
|
||||
updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, lastBasepullTime);
|
||||
insertCapturedDelete("4");
|
||||
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryIndexRow().getValueInstant("lastBasepullTime")).isEqualTo(lastBasepullTime);
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement()
|
||||
.satisfies(event -> assertThat(event.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: recovery hands over only the captured rows that existed when it
|
||||
** read them; a row captured after (by a run that started meanwhile) stays
|
||||
** AWAITING_REINDEX for that run.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_handsOverOnlyRowsCapturedBeforeItRead() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120);
|
||||
insertCapturedDelete("4");
|
||||
|
||||
AtomicBoolean capturedMeanwhile = new AtomicBoolean(false);
|
||||
new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected void updateIndexRow(Integer indexId, Map<String, ? extends Serializable> values) throws QException
|
||||
{
|
||||
super.updateIndexRow(indexId, values);
|
||||
if(capturedMeanwhile.compareAndSet(false, true))
|
||||
{
|
||||
insertCapturedDelete("5");
|
||||
}
|
||||
}
|
||||
}.recoverDeadRun(true);
|
||||
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("recordId") + " " + event.getValueString("status"))
|
||||
.containsExactlyInAnyOrder("4 " + QuickSearchFailedEvent.STATUS_PENDING, "5 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a recovery that fails while putting a table back leaves the dead
|
||||
** run's record RUNNING, so the next recovery still finds the run's start
|
||||
** and rewinds the basepull watermark.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_failsPartway_nextRecoveryStillRewinds() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, Instant.now().minus(5, ChronoUnit.MINUTES));
|
||||
QRecord deadRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120);
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected void updateIndexRow(Integer indexId, Map<String, ? extends Serializable> values) throws QException
|
||||
{
|
||||
throw (new QException("database unavailable"));
|
||||
}
|
||||
}.recoverDeadRun(true))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("database unavailable");
|
||||
|
||||
assertThat(queryRun(deadRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING);
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryIndexRow().getValueInstant("lastBasepullTime")).isEqualTo(deadRun.getValueInstant("startTime"));
|
||||
assertThat(queryRun(deadRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a run whose drain of captured deletes is so slow that it is judged
|
||||
** stopped, while a new full reindex starts, marks the table and captures a
|
||||
** delete of its own: the drain stops at its next batch and leaves both its
|
||||
** remaining row and the new run's row alone.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testDrainTakenOverMidway_leavesTheNewRunsRowsAlone() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(1);
|
||||
insertTestEntities(1);
|
||||
insertCapturedDelete("8");
|
||||
insertCapturedDelete("9");
|
||||
|
||||
AtomicBoolean newRunStarted = new AtomicBoolean(false);
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
if(newRunStarted.compareAndSet(false, true))
|
||||
{
|
||||
for(QRecord run : queryAll(QuickSearchIndexRun.TABLE_NAME))
|
||||
{
|
||||
backdateRun(run.getValueInteger("id"));
|
||||
}
|
||||
assertThat(new FullReindexStep().recoverDeadRun(false)).isEmpty();
|
||||
insertRunningRun(queryIndexRow().getValueInteger("id"), 0, 0);
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
insertCapturedDelete("7");
|
||||
}
|
||||
return (new BulkIndexResult().withSuccessCount(1));
|
||||
});
|
||||
|
||||
new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt());
|
||||
verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":8")), anyInt());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("recordId") + " " + event.getValueString("status"))
|
||||
.containsExactlyInAnyOrder("9 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX, "7 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the drain applies only rows up to the newest one at its start; a
|
||||
** row captured while it runs is left alone.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testDrain_stopsAtTheNewestRowReadAtItsStart() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(1);
|
||||
insertTestEntities(1);
|
||||
insertCapturedDelete("8");
|
||||
insertCapturedDelete("9");
|
||||
|
||||
AtomicBoolean capturedMeanwhile = new AtomicBoolean(false);
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
if(capturedMeanwhile.compareAndSet(false, true))
|
||||
{
|
||||
insertCapturedDelete("7");
|
||||
}
|
||||
return (new BulkIndexResult().withSuccessCount(1));
|
||||
});
|
||||
|
||||
new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
verify(mockClient, times(2)).deleteDocuments(anyList(), anyInt());
|
||||
assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement()
|
||||
.satisfies(event -> assertThat(event.getValueString("recordId")).isEqualTo("7"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: when the stale run records cannot be marked FAILED (every update
|
||||
** comes back with a per-row error), recovery fails and the new full
|
||||
** reindex does not start, since a live run misjudged as stopped would not
|
||||
** learn it was taken over.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_failedRunRecordsNotWritten_newRunDoesNotStart() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
insertIndexRow(true);
|
||||
updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
Integer staleRunId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id");
|
||||
|
||||
try(MockedConstruction<UpdateAction> updateActions = mockConstruction(UpdateAction.class, (updateAction, context) ->
|
||||
when(updateAction.execute(any())).thenAnswer(invocation ->
|
||||
{
|
||||
UpdateInput updateInput = invocation.getArgument(0);
|
||||
UpdateOutput updateOutput = new UpdateOutput();
|
||||
updateOutput.setRecords(updateInput.getRecords().stream().map(record -> new QRecord(record).withError(new SystemErrorStatusMessage("row is locked"))).toList());
|
||||
return (updateOutput);
|
||||
})))
|
||||
{
|
||||
assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("FAILED")
|
||||
.hasMessageContaining("row is locked");
|
||||
}
|
||||
|
||||
verify(mockClient, never()).createPhysicalIndex(anyString());
|
||||
verify(mockClient, never()).indexDocuments(anyString(), anyList(), anyInt());
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement()
|
||||
.satisfies(run ->
|
||||
{
|
||||
assertThat(run.getValueInteger("id")).isEqualTo(staleRunId);
|
||||
assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the run keeps itself alive at every point where it could otherwise
|
||||
** go quiet: after each page, right before its pre-swap ownership check,
|
||||
** before each table's update after the swap, and before each batch of the
|
||||
** captured-delete drain.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testAllTables_heartbeatsBeforeTheOwnershipCheckEachTableAndEachDrainBatch() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(1);
|
||||
insertTestEntities(1);
|
||||
insertCapturedDelete("8");
|
||||
insertCapturedDelete("9");
|
||||
|
||||
List<String> events = new ArrayList<>();
|
||||
doAnswer(invocation ->
|
||||
{
|
||||
events.add("refresh");
|
||||
return (null);
|
||||
}).when(mockClient).refreshIndex(PHYSICAL_INDEX);
|
||||
doAnswer(invocation ->
|
||||
{
|
||||
events.add("swap");
|
||||
return (null);
|
||||
}).when(mockClient).swapAliasTo(PHYSICAL_INDEX);
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
events.add("drainBatch");
|
||||
return (new BulkIndexResult().withSuccessCount(1));
|
||||
});
|
||||
|
||||
new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected void heartbeat(QuickSearchIndexRun run)
|
||||
{
|
||||
events.add("heartbeat");
|
||||
super.heartbeat(run);
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Override
|
||||
protected void updateIndexRow(Integer indexId, Map<String, ? extends Serializable> values) throws QException
|
||||
{
|
||||
if(values.containsKey("lastFullReindexTime"))
|
||||
{
|
||||
events.add("tableDone");
|
||||
}
|
||||
super.updateIndexRow(indexId, values);
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
assertThat(events).containsExactly(
|
||||
"heartbeat",
|
||||
"refresh", "heartbeat", "swap",
|
||||
"heartbeat", "tableDone",
|
||||
"heartbeat", "drainBatch", "heartbeat", "drainBatch", "heartbeat");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: when a run fails and cannot read whether a recovery took it over,
|
||||
** it leaves its run records RUNNING and its tables REBUILDING, so the next
|
||||
** recovery (once the records are stale) puts everything back, watermark
|
||||
** rewind included.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testFailure_takeoverUnreadable_leavesRunRecordsRunningForRecovery() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
AtomicBoolean databaseDown = new AtomicBoolean(false);
|
||||
when(mockClient.indexDocuments(anyString(), anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
databaseDown.set(true);
|
||||
throw (new QException("bulk failed"));
|
||||
});
|
||||
|
||||
assertThatThrownBy(() -> new FullReindexStep()
|
||||
{
|
||||
@Override
|
||||
protected QuickSearchQBitConfig getConfig() throws QException
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////
|
||||
// only the first read after the failure (the takeover check) is //
|
||||
// lost; everything after it would work //
|
||||
///////////////////////////////////////////////////////////////////
|
||||
if(databaseDown.getAndSet(false))
|
||||
{
|
||||
throw (new QException("database unavailable"));
|
||||
}
|
||||
return (super.getConfig());
|
||||
}
|
||||
}.run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("bulk failed");
|
||||
|
||||
verify(mockClient).deletePhysicalIndex(PHYSICAL_INDEX);
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
|
||||
QRecord run = queryAll(QuickSearchIndexRun.TABLE_NAME).get(0);
|
||||
assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING);
|
||||
|
||||
backdateRun(run.getValueInteger("id"));
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryRun(run.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: with nothing left behind, recovery changes nothing.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRecovery_nothingLeftBehind_changesNothing() throws QException
|
||||
{
|
||||
insertIndexRow(true);
|
||||
|
||||
assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty();
|
||||
|
||||
assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
|
||||
assertThat(queryIndexRow().getValueString("lastRunStatus")).isNull();
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).isEmpty();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in new issue
Block a user