Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 forge-recovery
|
a709dd5f47
|
||
|
|
ed0ab82b07
|
||
|
|
9004713636
|
||
 james.maesandCodex
|
5f9d5f47d8 | ||
|
james.maesandCodex
|
482ff5341b |
No files matched your search
+11
-68
@@ -1,71 +1,14 @@
|
||||
version: 2.1
|
||||
|
||||
setup: true
|
||||
orbs:
|
||||
qqq-orb: kingsrook/qqq-orb@0.5.2
|
||||
|
||||
commands:
|
||||
configure-package-install:
|
||||
steps:
|
||||
- run:
|
||||
name: Keep package installation unattended
|
||||
command: |
|
||||
# These build machines are discarded after the job.
|
||||
sudo install -d /etc/needrestart/conf.d
|
||||
printf '%s\n' '$nrconf{restart} = "l";' | sudo tee /etc/needrestart/conf.d/99-qqq-ci.conf >/dev/null
|
||||
|
||||
qqq-orb: kingsrook/qqq-orb@0.8
|
||||
workflows:
|
||||
test_only:
|
||||
jobs:
|
||||
- qqq-orb/mvn_test_only:
|
||||
pre-steps:
|
||||
- configure-package-install
|
||||
context: [qqq-maven-registry-credentials]
|
||||
filters:
|
||||
branches:
|
||||
ignore: /(develop|main|release\/.*|hotfix\/.*)/
|
||||
|
||||
publish_snapshot:
|
||||
jobs:
|
||||
- qqq-orb/mvn_publish:
|
||||
pre-steps:
|
||||
- configure-package-install
|
||||
context: [qqq-maven-registry-credentials]
|
||||
branch_type: snapshot
|
||||
filters:
|
||||
branches:
|
||||
only: [develop]
|
||||
|
||||
publish_release_candidate:
|
||||
jobs:
|
||||
- qqq-orb/mvn_publish:
|
||||
pre-steps:
|
||||
- configure-package-install
|
||||
context: [qqq-maven-registry-credentials]
|
||||
branch_type: release_candidate
|
||||
filters:
|
||||
branches:
|
||||
only: [/release\/.*/]
|
||||
|
||||
publish_release:
|
||||
jobs:
|
||||
- qqq-orb/mvn_publish:
|
||||
pre-steps:
|
||||
- configure-package-install
|
||||
context: [qqq-maven-registry-credentials]
|
||||
branch_type: release
|
||||
filters:
|
||||
branches:
|
||||
only: [main]
|
||||
tags:
|
||||
only: [/v.*/]
|
||||
|
||||
publish_hotfix_release:
|
||||
jobs:
|
||||
- qqq-orb/mvn_publish:
|
||||
pre-steps:
|
||||
- configure-package-install
|
||||
context: [qqq-maven-registry-credentials]
|
||||
branch_type: hotfix
|
||||
filters:
|
||||
branches:
|
||||
only: [/hotfix\/.*/]
|
||||
setup:
|
||||
jobs:
|
||||
- qqq-orb/generate_pipeline:
|
||||
orb_version: '0.8'
|
||||
filters:
|
||||
branches:
|
||||
ignore: automation/qbit-rc-pending
|
||||
tags:
|
||||
only: /v.*/
|
||||
@@ -0,0 +1,63 @@
|
||||
{
|
||||
"permissions": {
|
||||
"allow": [
|
||||
"Bash(find:*)",
|
||||
"Bash(grep:*)",
|
||||
"Bash(mvn compile:*)",
|
||||
"Bash(jar -tf:*)",
|
||||
"Bash(jar -xf:*)",
|
||||
"Bash(cat:*)",
|
||||
"Bash(mvn test:*)",
|
||||
"Bash(done)",
|
||||
"Bash(mvn verify:*)",
|
||||
"Bash(git add:*)",
|
||||
"Bash(git commit:*)",
|
||||
"Bash(git push:*)",
|
||||
"mcp__circleci-mcp-server__get_latest_pipeline_status",
|
||||
"mcp__circleci-mcp-server__list_followed_projects",
|
||||
"mcp__circleci-mcp-server__run_pipeline",
|
||||
"mcp__circleci-mcp-server__get_build_failure_logs",
|
||||
"mcp__circleci-mcp-server__config_helper",
|
||||
"mcp__github__get_file_contents",
|
||||
"Bash(mvn dependency:tree:*)",
|
||||
"Bash(mvn help:effective-pom:*)",
|
||||
"Bash(git checkout:*)",
|
||||
"Bash(ls /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/docs/*.md)",
|
||||
"Bash(git:*)",
|
||||
"Bash(ls:*)",
|
||||
"Bash(open:*)",
|
||||
"Bash(WT=/Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/.claude/worktrees/agent-a9dc6e8e)",
|
||||
"Bash(DEST=/Users/james.maes/Git.Local/QRun-IO/qbit-quick-search)",
|
||||
"Bash(cp \"$WT/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/BulkIndexResult.java\" \"$DEST/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/\")",
|
||||
"Bash(mvn clean:*)",
|
||||
"Bash(jar tf:*)",
|
||||
"Bash(unzip -p /Users/james.maes/.m2/repository/com/kingsrook/qqq/qqq-backend-core/0.40.0-SNAPSHOT/qqq-backend-core-0.40.0-SNAPSHOT.jar com/kingsrook/qqq/backend/core/actions/customizers/TableCustomizerInterface.java)",
|
||||
"Bash(sed -i '' 's/postInsertRecord/postInsert/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java)",
|
||||
"Bash(sed -i '' 's/postUpdateRecord/postUpdate/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java)",
|
||||
"Bash(sed -i '' 's/postDeleteRecord/postDelete/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java)",
|
||||
"Bash(docker info:*)",
|
||||
"Bash(docker run:*)",
|
||||
"Read(//var/run/**)",
|
||||
"Read(//Users/james.maes/**)",
|
||||
"Bash(echo \"DOCKER_HOST=$DOCKER_HOST\")",
|
||||
"Bash(echo \"TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=$TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE\")",
|
||||
"Bash(DOCKER_HOST=unix:///var/run/docker.sock TESTCONTAINERS_RYUK_DISABLED=true mvn verify -Dit.test=OpenSearchIntegrationTest)",
|
||||
"Bash(docker version:*)",
|
||||
"Bash(DOCKER_HOST=unix:///var/run/docker.sock mvn verify -Dit.test=OpenSearchIntegrationTest)",
|
||||
"Bash(curl -s --unix-socket /var/run/docker.sock http://localhost/v1.24/info)",
|
||||
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} OS: {d.get\\(OperatingSystem,?\\)}''''\\)\")",
|
||||
"Bash(curl -s --unix-socket /Users/james.maes/Library/Containers/com.docker.docker/Data/docker-cli.sock http://localhost/v1.47/info)",
|
||||
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} OS: {d.get\\(OperatingSystem,?\\)} Containers: {d.get\\(Containers,?\\)}''''\\)\")",
|
||||
"Bash(curl -s --unix-socket /Users/james.maes/.docker/run/docker.sock http://localhost/v1.47/info)",
|
||||
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} NCPU: {d.get\\(NCPU,?\\)} OSType: {d.get\\(OSType,?\\)}''''\\)\")",
|
||||
"Bash(DOCKER_HOST=unix:///Users/james.maes/.docker/run/docker.sock DOCKER_API_VERSION=1.47 mvn verify -Dit.test=OpenSearchIntegrationTest -pl .)",
|
||||
"Bash(cp /Users/james.maes/Git.Local/QRun-IO/qbit-webhooks/LICENSE /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/LICENSE)",
|
||||
"Bash(cp -r /Users/james.maes/Git.Local/QRun-IO/qbit-webhooks/checkstyle/* /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/checkstyle/)",
|
||||
"Bash(gh pr:*)",
|
||||
"Bash(gh repo:*)",
|
||||
"Bash(gh api:*)",
|
||||
"Bash(python3 -c \":*)",
|
||||
"mcp__circleci-mcp-server__get_job_test_results"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
# Standard QRun release caller (v2, thin contract): every push and manual dispatch delegates to the protected qrun-release-v1 channel. Do not edit per repository.
|
||||
name: qrun-release
|
||||
on:
|
||||
push: {}
|
||||
workflow_dispatch: {}
|
||||
permissions:
|
||||
contents: read
|
||||
actions: read
|
||||
jobs:
|
||||
release:
|
||||
uses: self:KofTwentyTwo/platform-ci/.gitea/workflows/java-docker.yaml@qrun-release-v1
|
||||
with:
|
||||
runner: linux-small
|
||||
docker-runner: container-build
|
||||
browsers: true
|
||||
browser-runner: browser-build
|
||||
postpublish:
|
||||
if: ${{ needs.release.outputs.capture == 'true' }}
|
||||
needs: release
|
||||
permissions:
|
||||
contents: write
|
||||
actions: write
|
||||
uses: self:KofTwentyTwo/platform-ci/.gitea/workflows/qrun-source-version.yaml@qrun-release-v1
|
||||
with:
|
||||
runner: linux-small
|
||||
@@ -1 +0,0 @@
|
||||
* @KofTwentyTwo
|
||||
@@ -0,0 +1,23 @@
|
||||
name: CodeQL
|
||||
'on':
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
- develop
|
||||
- release/**
|
||||
- hotfix/**
|
||||
pull_request: {}
|
||||
schedule:
|
||||
- cron: 41 16 * * 0
|
||||
workflow_dispatch: {}
|
||||
permissions:
|
||||
contents: read
|
||||
actions: read
|
||||
security-events: write
|
||||
packages: read
|
||||
concurrency:
|
||||
group: codeql-${{ github.event.pull_request.number || github.ref }}
|
||||
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
|
||||
jobs:
|
||||
analyze:
|
||||
uses: QRun-IO/qqq-orb/.github/workflows/qbit-codeql.yml@c288a8864fe9054f68410e1d78f5863aea4c4d22
|
||||
@@ -18,4 +18,3 @@ hs_err_pid*
|
||||
dependency-reduced-pom.xml
|
||||
/.env.local
|
||||
/.cache/
|
||||
.claude/settings.local.json
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
profile: qbit
|
||||
orb_version: '0.8'
|
||||
default_branch: develop
|
||||
java_version: '21'
|
||||
browsers: true
|
||||
api_checks: false
|
||||
docs_only: false
|
||||
release_on_main: true
|
||||
release_tag_policy: v-prefix
|
||||
test_branch_policy: non-release-no-main-hotfix
|
||||
publish:
|
||||
snapshot: true
|
||||
release_candidate: true
|
||||
release: true
|
||||
hotfix: true
|
||||
asciidoc: false
|
||||
@@ -4,22 +4,20 @@ This file provides guidance to Claude Code when working with code in this reposi
|
||||
|
||||
## Project Overview
|
||||
|
||||
**QBit Quick Search** - A QQQ QBit providing OpenSearch-backed full-text search across QQQ applications.
|
||||
**QBit Quick Search** - A QQQ QBit providing OpenSearch-backed full-text search across QQQ applications. Current version: 0.1.0.
|
||||
|
||||
## Build Commands
|
||||
|
||||
```bash
|
||||
mvn compile # Compile
|
||||
mvn test # Run unit tests
|
||||
mvn test # Run all tests (174 tests)
|
||||
mvn test -Dtest=ClassName # Run single test class
|
||||
mvn package # Build JAR
|
||||
mvn verify # Full verify, including integration tests
|
||||
mvn verify -Pqqq-snapshot # Verify against 4.1.0-SNAPSHOT (override with -Dqqq.snapshot.version)
|
||||
mvn verify # Full verify with coverage check
|
||||
```
|
||||
|
||||
Integration tests (require Docker) run via `mvn verify` using maven-failsafe-plugin.
|
||||
Unit tests exclude `*IntegrationTest.java` and `*IT.java`.
|
||||
Integration tests use `@ExtendWith(RequiresDockerCondition.class)`: skipped without Docker locally, failed when `CI=true`.
|
||||
|
||||
## Architecture
|
||||
|
||||
@@ -34,14 +32,13 @@ Integration tests use `@ExtendWith(RequiresDockerCondition.class)`: skipped with
|
||||
|
||||
**Customizers** (`customizers/` package, all three registered when `enableRealTimeIndexing = true`):
|
||||
- `QuickSearchPostInsertCustomizer` - Fires after insert, publishes index events
|
||||
- `QuickSearchPostUpdateCustomizer` - Fires after update, re-reads the full records from the source table, publishes index events
|
||||
- `QuickSearchPostUpdateCustomizer` - Fires after update, publishes index events
|
||||
- `QuickSearchPostDeleteCustomizer` - Fires after delete, publishes delete events
|
||||
|
||||
**Processes** (`processes/` package):
|
||||
- `AbstractIndexingStep` - Base class with shared indexing logic (query source table, build documents, publish events, update run record)
|
||||
- `BasepullIndexStep` - Queries records modified since last run timestamp
|
||||
- `FullReindexStep` - Wipes and rebuilds the index for one or all tables
|
||||
- `ReconcileIndexStep` - Re-indexes each table in primary-key order, then removes documents with no source record (no wipe, so no search blackout)
|
||||
- `IndexingUtils` - Converts `QRecord` values to indexed text using field configs
|
||||
|
||||
**Publisher** (`publisher/` package):
|
||||
@@ -71,18 +68,9 @@ Integration tests use `@ExtendWith(RequiresDockerCondition.class)`: skipped with
|
||||
## Dependencies
|
||||
|
||||
- Java 21
|
||||
- Parent `com.kingsrook:qbit-build-parent:2.0.0` (pins QQQ 4.0.0). Release 0.2.1 deviates from ADR-0007: the `qqq.version` property (4.1.0-RC.1) drives a `qqq-bom-pom` import that replaces the parent's BOM; remove it once a parent release on the QQQ 4.1 line exists. The opt-in `qqq-snapshot` profile (`-Pqqq-snapshot`) imports `qqq-bom-pom:${qqq.snapshot.version}` (default 4.1.0-SNAPSHOT) instead and adds the Central snapshots repository
|
||||
- QQQ Backend Core 0.40.0-SNAPSHOT (via `qqq-bom-pom`)
|
||||
- OpenSearch Java Client 2.10.0
|
||||
- Apache HttpClient5 and jackson-datatype-jsr310, versions managed by the qqq BOM (keep them unpinned so they match its httpcore5 and jackson)
|
||||
- Apache HttpClient5 5.3
|
||||
- jackson-datatype-jsr310 2.21.0 (Instant serialization for OpenSearch client)
|
||||
- JUnit 5 + AssertJ + Mockito for testing
|
||||
- Testcontainers 1.21.4 for integration tests (requires Docker)
|
||||
|
||||
## Knowledge base
|
||||
|
||||
Second-brain vault notes covering this repo and the QQQ platform it plugs into:
|
||||
|
||||
- Platform hub: `$SECOND_BRAIN_VAULT/knowledge/qqq/qqq-hub.md` (start here; read `knowledge/qqq/architecture/metadata-model.md` for QBit mechanics)
|
||||
- This repo's dossier: `$SECOND_BRAIN_VAULT/knowledge/qqq/repos/qbit-quick-search.md` (reviewed at develop @ `8d9ec2711fde`, 2026-07-04)
|
||||
- Production-readiness audit + P0-P3 roadmap: `$SECOND_BRAIN_VAULT/projects/qbit-quick-search.md`
|
||||
|
||||
Note: the dossier was reviewed before the qqq 4.0 re-pin; trust this file for build and dependency facts.
|
||||
- Testcontainers for integration tests (requires Docker)
|
||||
@@ -1,6 +1,6 @@
|
||||
# QBit: Quick Search
|
||||
|
||||
[](https://github.com/QRun-IO/qbit-quick-search)
|
||||
[](https://github.com/QRun-IO/qbit-quick-search)
|
||||
[](https://www.apache.org/licenses/LICENSE-2.0)
|
||||
[](https://adoptium.net/)
|
||||
|
||||
@@ -35,7 +35,7 @@ QBit Quick Search is 100% open source under Apache 2.0. All data stays in your O
|
||||
|
||||
### Technology Stack
|
||||
|
||||
- **Java 21** with QQQ 4.x backend modules (QQQ 4.1.0-RC.1 BOM, build plugins from `qbit-build-parent` 2.0.0)
|
||||
- **Java 21** with QQQ backend modules
|
||||
- **OpenSearch 2.x** for full-text search with custom edge-ngram analyzer
|
||||
- **QQQ Framework**: Entities, processes, customizers, permissions, API layer
|
||||
|
||||
@@ -54,7 +54,7 @@ qbit-quick-search/
|
||||
model/ -- QuickSearchIndex, QuickSearchIndexRun entities
|
||||
opensearch/ -- OpenSearch client, document model, bulk results
|
||||
actions/ -- QuickSearchAction, input/output DTOs
|
||||
processes/ -- AbstractIndexingStep, BasepullIndexStep, FullReindexStep, ReconcileIndexStep
|
||||
processes/ -- AbstractIndexingStep, BasepullIndexStep, FullReindexStep
|
||||
customizers/ -- Post-insert, post-update, post-delete customizers
|
||||
publisher/ -- IndexEventPublisher interface, SynchronousIndexEventPublisher
|
||||
```
|
||||
@@ -76,7 +76,7 @@ qbit-quick-search/
|
||||
<dependency>
|
||||
<groupId>com.kingsrook.qbits</groupId>
|
||||
<artifactId>qbit-quick-search</artifactId>
|
||||
<version>0.2.1</version>
|
||||
<version>0.2.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
@@ -253,24 +253,6 @@ new FullReindexStep().run(input, output);
|
||||
|
||||
Or through the QQQ admin UI via the "Full Reindex" process.
|
||||
|
||||
### Reconciling the Index
|
||||
|
||||
If the index may have drifted from the source tables (for example, deletes that
|
||||
happened while OpenSearch was down), run the reconcile process. It re-indexes
|
||||
every source record, then removes documents whose source record no longer
|
||||
exists. It does not wipe the index first, so search keeps working while it runs.
|
||||
|
||||
```java
|
||||
RunBackendStepInput input = new RunBackendStepInput();
|
||||
input.addValue("tableName", "customer"); // optional: omit to reconcile all tables
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new ReconcileIndexStep().run(input, output);
|
||||
// output values: recordsIndexed, documentsRemoved
|
||||
```
|
||||
|
||||
Or through the QQQ admin UI via the "Reconcile Index" process. If any document
|
||||
fails to index, the run is marked `FAILED` and no documents are removed.
|
||||
|
||||
## Data Model
|
||||
|
||||
### Tables (2)
|
||||
@@ -280,13 +262,12 @@ fails to index, the run is marked `FAILED` and no documents are removed.
|
||||
| `quickSearchIndex` | Per-table index configuration, status, and basepull tracking |
|
||||
| `quickSearchIndexRun` | Individual indexing run history with record counts and error tracking |
|
||||
|
||||
### Processes (3)
|
||||
### Processes (2)
|
||||
|
||||
| Process | Description |
|
||||
|---------|-------------|
|
||||
| Basepull Index | Scheduled: queries each source table for records modified since last run, indexes in batches |
|
||||
| Full Reindex | On-demand: deletes and re-indexes all records for one or all tables |
|
||||
| Reconcile Index | On-demand: re-indexes all records for one or all tables, then removes documents with no source record, without wiping the index first |
|
||||
|
||||
## Configuration Reference
|
||||
|
||||
@@ -343,21 +324,11 @@ Index `wmsItem` and `wmsLocation` for warehouse item and location search.
|
||||
## Testing
|
||||
|
||||
```bash
|
||||
mvn test # Run unit tests
|
||||
mvn test # Run all 174 unit tests
|
||||
mvn verify # Run unit tests + integration tests (requires Docker)
|
||||
mvn test -Dtest=ClassName # Run single test class
|
||||
mvn verify -Pqqq-snapshot # Verify against the next qqq line (4.1.0-SNAPSHOT)
|
||||
```
|
||||
|
||||
The qqq version comes from the `qqq.version` property (4.1.0-RC.1), whose
|
||||
`qqq-bom-pom` import replaces the QQQ 4.0.0 BOM in `qbit-build-parent` 2.0.0.
|
||||
The opt-in `qqq-snapshot` profile imports `qqq-bom-pom` at
|
||||
`${qqq.snapshot.version}` (default `4.1.0-SNAPSHOT`) instead and adds the
|
||||
Central snapshots repository. Pick another version with `-Dqqq.snapshot.version=...`.
|
||||
|
||||
Without Docker, integration tests are skipped locally. With `CI=true`, they fail
|
||||
instead, so a CI build cannot pass with its integration tests silently skipped.
|
||||
|
||||
### Coverage
|
||||
|
||||
- **70%+ instruction coverage** (JaCoCo enforced)
|
||||
|
||||
@@ -0,0 +1,288 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2021-2024. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as
|
||||
* published by the Free Software Foundation, either version 3 of the
|
||||
* License, or (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qqq.backend.core.actions.customizers;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.QueryOrGetInputInterface;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Common interface used by all (core) TableCustomizer types (e.g., post-query,
|
||||
** and {pre,post}-{insert,update,delete}.
|
||||
**
|
||||
** Note that the abstract-base classes for each action still exist, though have
|
||||
** been back-ported to be implementors of this interface. The action classes
|
||||
** will now expect this type, and call this type's methods.
|
||||
**
|
||||
*******************************************************************************/
|
||||
public interface TableCustomizerInterface
|
||||
{
|
||||
QLogger LOG = QLogger.getLogger(TableCustomizerInterface.class);
|
||||
|
||||
/*******************************************************************************
|
||||
** custom actions to run after a query (or get!) takes place.
|
||||
**
|
||||
*******************************************************************************/
|
||||
default List<QRecord> postQuery(QueryOrGetInputInterface queryInput, List<QRecord> records) throws QException
|
||||
{
|
||||
LOG.info("A default implementation of postQuery is running... Probably not expected!", logPair("tableName", queryInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** custom actions before an insert takes place.
|
||||
**
|
||||
** It's important for implementations to be aware of the isPreview field, which
|
||||
** is set to true when the code is running to give users advice, e.g., on a review
|
||||
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
|
||||
** things like storing data, you don't want to do that if isPreview is true!!
|
||||
**
|
||||
** General implementation would be, to iterate over the records (the inputs to
|
||||
** the insert action), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
|
||||
** - possibly manipulating values (`setValue`)
|
||||
** - possibly throwing an exception - if you really don't want the insert operation to continue.
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) that you want to go on to the backend implementation class.
|
||||
*******************************************************************************/
|
||||
default List<QRecord> preInsert(InsertInput insertInput, List<QRecord> records, boolean isPreview) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
return (preInsertOrUpdate(insertInput, records, isPreview, Optional.empty()));
|
||||
}
|
||||
catch(NotImplementedHereException e)
|
||||
{
|
||||
LOG.info("A default implementation of preInsert is running... Probably not expected!", logPair("tableName", insertInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
default AbstractPreInsertCustomizer.WhenToRun whenToRunPreInsert(InsertInput insertInput, boolean isPreview)
|
||||
{
|
||||
return (AbstractPreInsertCustomizer.WhenToRun.AFTER_ALL_VALIDATIONS);
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** custom actions after an insert takes place.
|
||||
**
|
||||
** General implementation would be, to iterate over the records (the outputs of
|
||||
** the insert action), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
|
||||
** - possibly throwing an exception - though doing so won't stop the update, and instead
|
||||
** will just set a warning on all of the updated records...
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) that you want to go back to the caller.
|
||||
*******************************************************************************/
|
||||
default List<QRecord> postInsert(InsertInput insertInput, List<QRecord> records) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
return (postInsertOrUpdate(insertInput, records, Optional.empty()));
|
||||
}
|
||||
catch(NotImplementedHereException e)
|
||||
{
|
||||
LOG.info("A default implementation of postInsert is running... Probably not expected!", logPair("tableName", insertInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** custom actions before an update takes place.
|
||||
**
|
||||
** It's important for implementations to be aware of the isPreview field, which
|
||||
** is set to true when the code is running to give users advice, e.g., on a review
|
||||
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
|
||||
** things like storing data, you don't want to do that if isPreview is true!!
|
||||
**
|
||||
** General implementation would be, to iterate over the records (the inputs to
|
||||
** the update action), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
|
||||
** - possibly manipulating values (`setValue`)
|
||||
** - possibly throwing an exception - if you really don't want the update operation to continue.
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) that you want to go on to the backend implementation class.
|
||||
**
|
||||
** Note, "old records" (e.g., with values freshly fetched from the backend) will be
|
||||
** available (if the backend supports it)
|
||||
*******************************************************************************/
|
||||
default List<QRecord> preUpdate(UpdateInput updateInput, List<QRecord> records, boolean isPreview, Optional<List<QRecord>> oldRecordList) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
return (preInsertOrUpdate(updateInput, records, isPreview, oldRecordList));
|
||||
}
|
||||
catch(NotImplementedHereException e)
|
||||
{
|
||||
LOG.info("A default implementation of preUpdate is running... Probably not expected!", logPair("tableName", updateInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** custom actions after an update takes place.
|
||||
**
|
||||
** General implementation would be, to iterate over the records (the outputs of
|
||||
** the update action), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records?
|
||||
** - possibly throwing an exception - though doing so won't stop the update, and instead
|
||||
** will just set a warning on all of the updated records...
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) that you want to go back to the caller.
|
||||
**
|
||||
** Note, "old records" (e.g., with values freshly fetched from the backend) will be
|
||||
** available (if the backend supports it).
|
||||
*******************************************************************************/
|
||||
default List<QRecord> postUpdate(UpdateInput updateInput, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
return (postInsertOrUpdate(updateInput, records, oldRecordList));
|
||||
}
|
||||
catch(NotImplementedHereException e)
|
||||
{
|
||||
LOG.info("A default implementation of postUpdate is running... Probably not expected!", logPair("tableName", updateInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Custom actions before a delete takes place.
|
||||
**
|
||||
** It's important for implementations to be aware of the isPreview param, which
|
||||
** is set to true when the code is running to give users advice, e.g., on a review
|
||||
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
|
||||
** things like storing data, you don't want to do that if isPreview is true!!
|
||||
**
|
||||
** General implementation would be, to iterate over the records (which the DeleteAction
|
||||
** would look up based on the inputs to the delete action), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
|
||||
** - possibly throwing an exception - if you really don't want the delete operation to continue.
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) - this is how errors
|
||||
** and warnings are propagated to the DeleteAction. Note that any records with
|
||||
** an error will NOT proceed to the backend's delete interface - but those with
|
||||
** warnings will.
|
||||
*******************************************************************************/
|
||||
default List<QRecord> preDelete(DeleteInput deleteInput, List<QRecord> records, boolean isPreview) throws QException
|
||||
{
|
||||
LOG.info("A default implementation of preDelete is running... Probably not expected!", logPair("tableName", deleteInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Custom actions after a delete takes place.
|
||||
**
|
||||
** General implementation would be, to iterate over the records (ones which didn't
|
||||
** have a delete error), and look at their values:
|
||||
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records?
|
||||
** - possibly throwing an exception - though doing so won't stop the delete, and instead
|
||||
** will just set a warning on all of the deleted records...
|
||||
** - doing "whatever else" you may want to do.
|
||||
** - returning the list of records (can be the input list) that you want to go back
|
||||
** to the caller - this is how errors and warnings are propagated .
|
||||
*******************************************************************************/
|
||||
default List<QRecord> postDelete(DeleteInput deleteInput, List<QRecord> records) throws QException
|
||||
{
|
||||
LOG.info("A default implementation of postDelete is running... Probably not expected!", logPair("tableName", deleteInput.getTableName()));
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Optional method to override in a customizer that does the same thing for
|
||||
** both preInsert & preUpdate.
|
||||
***************************************************************************/
|
||||
default List<QRecord> preInsertOrUpdate(AbstractActionInput input, List<QRecord> records, boolean isPreview, Optional<List<QRecord>> oldRecordList) throws QException
|
||||
{
|
||||
throw NotImplementedHereException.instance;
|
||||
}
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Optional method to override in a customizer that does the same thing for
|
||||
** both postInsert & postUpdate.
|
||||
***************************************************************************/
|
||||
default List<QRecord> postInsertOrUpdate(AbstractActionInput input, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
|
||||
{
|
||||
throw NotImplementedHereException.instance;
|
||||
}
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
**
|
||||
***************************************************************************/
|
||||
default Optional<Map<Serializable, QRecord>> oldRecordListToMap(String primaryKeyField, Optional<List<QRecord>> oldRecordList)
|
||||
{
|
||||
if(oldRecordList.isPresent())
|
||||
{
|
||||
return (Optional.of(CollectionUtils.listToMap(oldRecordList.get(), r -> r.getValue(primaryKeyField))));
|
||||
}
|
||||
else
|
||||
{
|
||||
return (Optional.empty());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
**
|
||||
***************************************************************************/
|
||||
class NotImplementedHereException extends QException
|
||||
{
|
||||
private static NotImplementedHereException instance = new NotImplementedHereException();
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
**
|
||||
***************************************************************************/
|
||||
private NotImplementedHereException()
|
||||
{
|
||||
super("Not implemented here");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2021-2022. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as
|
||||
* published by the Free Software Foundation, either version 3 of the
|
||||
* License, or (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qqq.backend.core.model.metadata;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Interface to be implemented by enums of the possible supplemental types
|
||||
**
|
||||
*******************************************************************************/
|
||||
public interface SupplementalCustomizerType
|
||||
{
|
||||
/*******************************************************************************
|
||||
** allows implementing enums to return the expected class being used
|
||||
*******************************************************************************/
|
||||
public Class<?> getCodeReferenceType();
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -18,17 +18,16 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<properties>
|
||||
<revision>0.2.1</revision>
|
||||
<revision>0.2.0</revision>
|
||||
<maven.compiler.source>21</maven.compiler.source>
|
||||
<maven.compiler.target>21</maven.compiler.target>
|
||||
<opensearch.version>2.10.0</opensearch.version>
|
||||
<qqq.version>4.1.0-RC.1</qqq.version>
|
||||
</properties>
|
||||
|
||||
<parent>
|
||||
<groupId>com.kingsrook</groupId>
|
||||
<artifactId>qbit-build-parent</artifactId>
|
||||
<version>2.0.0</version>
|
||||
<version>1.6.0</version>
|
||||
</parent>
|
||||
|
||||
<groupId>com.kingsrook.qbits</groupId>
|
||||
@@ -76,13 +75,10 @@
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
<!-- Release 0.2.1 targets QQQ 4.1.0-RC.1. qbit-build-parent 2.0.0 still -->
|
||||
<!-- pins QQQ 4.0.0, so this import replaces the parent's qqq BOM until a -->
|
||||
<!-- parent release on the QQQ 4.1 line exists (deviation from ADR-0007). -->
|
||||
<dependency>
|
||||
<groupId>com.kingsrook.qqq</groupId>
|
||||
<artifactId>qqq-bom-pom</artifactId>
|
||||
<version>${qqq.version}</version>
|
||||
<version>0.40.0</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
@@ -101,16 +97,16 @@
|
||||
<version>${opensearch.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- httpclient5 and jackson versions are managed by the qqq BOM, so they -->
|
||||
<!-- stay binary-compatible with the httpcore5 and jackson it brings. -->
|
||||
<dependency>
|
||||
<groupId>org.apache.httpcomponents.client5</groupId>
|
||||
<artifactId>httpclient5</artifactId>
|
||||
<version>5.3</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jsr310</artifactId>
|
||||
<version>2.21.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
@@ -151,13 +147,13 @@
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers</artifactId>
|
||||
<version>1.21.4</version>
|
||||
<version>1.21.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<version>1.21.4</version>
|
||||
<version>1.21.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
@@ -228,39 +224,4 @@
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
<profiles>
|
||||
<!-- Opt-in check against the next qqq line: mvn -B verify -Pqqq-snapshot -->
|
||||
<!-- (override the version with -Dqqq.snapshot.version=...). Default builds -->
|
||||
<!-- take the qqq version from the qqq.version property above. -->
|
||||
<profile>
|
||||
<id>qqq-snapshot</id>
|
||||
<properties>
|
||||
<qqq.snapshot.version>4.1.0-SNAPSHOT</qqq.snapshot.version>
|
||||
</properties>
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.kingsrook.qqq</groupId>
|
||||
<artifactId>qqq-bom-pom</artifactId>
|
||||
<version>${qqq.snapshot.version}</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>sonatype-snapshots</id>
|
||||
<url>https://central.sonatype.com/repository/maven-snapshots/</url>
|
||||
<releases>
|
||||
<enabled>false</enabled>
|
||||
</releases>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
</repository>
|
||||
</repositories>
|
||||
</profile>
|
||||
</profiles>
|
||||
</project>
|
||||
@@ -43,7 +43,6 @@ import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import com.kingsrook.qbits.quicksearch.processes.BasepullIndexStep;
|
||||
import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
|
||||
import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher;
|
||||
|
||||
@@ -63,7 +62,6 @@ public class QuickSearchQBitProducer
|
||||
|
||||
public static final String BASEPULL_PROCESS_NAME = "quickSearchBasepullIndex";
|
||||
public static final String FULL_REINDEX_PROCESS_NAME = "quickSearchFullReindex";
|
||||
public static final String RECONCILE_PROCESS_NAME = "quickSearchReconcileIndex";
|
||||
public static final String APP_NAME = "quickSearchAdmin";
|
||||
|
||||
private QuickSearchQBitConfig config;
|
||||
@@ -219,17 +217,14 @@ public class QuickSearchQBitProducer
|
||||
////////////////////////////////////////////////////
|
||||
String basepullProcessName = config.applyPrefix(BASEPULL_PROCESS_NAME);
|
||||
String fullReindexProcessName = config.applyPrefix(FULL_REINDEX_PROCESS_NAME);
|
||||
String reconcileProcessName = config.applyPrefix(RECONCILE_PROCESS_NAME);
|
||||
|
||||
QProcessMetaData basepullProcess = buildBasepullProcess(basepullProcessName);
|
||||
QProcessMetaData fullReindexProcess = buildFullReindexProcess(fullReindexProcessName);
|
||||
QProcessMetaData reconcileProcess = buildReconcileProcess(reconcileProcessName);
|
||||
|
||||
if(Boolean.TRUE.equals(config.getEnableScheduledProcesses()))
|
||||
{
|
||||
qInstance.addProcess(basepullProcess);
|
||||
qInstance.addProcess(fullReindexProcess);
|
||||
qInstance.addProcess(reconcileProcess);
|
||||
}
|
||||
|
||||
////////////////////////////////////////////////////
|
||||
@@ -254,7 +249,6 @@ public class QuickSearchQBitProducer
|
||||
{
|
||||
app.withChild(basepullProcess);
|
||||
app.withChild(fullReindexProcess);
|
||||
app.withChild(reconcileProcess);
|
||||
}
|
||||
|
||||
qInstance.addApp(app);
|
||||
@@ -533,23 +527,6 @@ public class QuickSearchQBitProducer
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Build the reconcile process metadata: re-index each table from its
|
||||
** source and remove documents with no source record, without wiping the
|
||||
** index first. Takes the same optional tableName input as full reindex.
|
||||
***************************************************************************/
|
||||
private QProcessMetaData buildReconcileProcess(String processName)
|
||||
{
|
||||
return new QProcessMetaData()
|
||||
.withName(processName)
|
||||
.withLabel("Quick Search Reconcile Index")
|
||||
.withStep(new QBackendStepMetaData()
|
||||
.withName("reconcile")
|
||||
.withCode(new QCodeReference(ReconcileIndexStep.class)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Register post-insert, post-update, and post-delete customizers on each
|
||||
** discovered source table.
|
||||
|
||||
+4
-18
@@ -24,7 +24,6 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
|
||||
@@ -57,8 +56,8 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Post-delete hook: builds DELETE events for each successfully deleted
|
||||
** record and publishes them via the configured IndexEventPublisher.
|
||||
** Post-delete hook: builds DELETE events for each deleted record and publishes
|
||||
** them via the configured IndexEventPublisher.
|
||||
**
|
||||
** If the publisher is null (QBit not yet initialized) or if publishing fails,
|
||||
** a warning is logged and the original records are returned without throwing,
|
||||
@@ -88,26 +87,13 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
|
||||
List<IndexEvent> events = new ArrayList<>();
|
||||
for(QRecord record : records)
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////
|
||||
// a record with errors was not deleted, so its document stays; //
|
||||
// one without a primary key has no document id to remove //
|
||||
///////////////////////////////////////////////////////////////////
|
||||
String recordId = record.getValueString(primaryKeyField);
|
||||
if(recordId == null || CollectionUtils.nullSafeHasContents(record.getErrors()))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
events.add(new IndexEvent()
|
||||
.withAction(IndexEventAction.DELETE)
|
||||
.withTableName(tableName)
|
||||
.withRecordId(recordId));
|
||||
.withRecordId(record.getValueString(primaryKeyField)));
|
||||
}
|
||||
|
||||
if(!events.isEmpty())
|
||||
{
|
||||
publisher.publishDeleteEvents(events);
|
||||
}
|
||||
publisher.publishDeleteEvents(events);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
|
||||
+4
-53
@@ -17,21 +17,14 @@
|
||||
package com.kingsrook.qbits.quicksearch.customizers;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
|
||||
@@ -44,11 +37,6 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
** Post-update table customizer that re-indexes updated records in the Quick
|
||||
** Search OpenSearch index.
|
||||
**
|
||||
** An update's records may hold only the fields that changed, so each updated
|
||||
** record is re-read from the source table (in the update's transaction) and
|
||||
** the full record is indexed. Indexing the sparse record would drop the
|
||||
** unchanged fields from the document.
|
||||
**
|
||||
** QQQ instantiates this class via reflection using the no-arg constructor.
|
||||
** The publisher and table configuration are obtained from the static
|
||||
** QuickSearchQBitContext at call time.
|
||||
@@ -69,10 +57,8 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Post-update hook: re-reads the successfully updated records from the
|
||||
** source table, builds INDEX events from them, and publishes them via the
|
||||
** configured IndexEventPublisher. Records with errors were not updated, so
|
||||
** they are skipped.
|
||||
** Post-update hook: builds INDEX events for each updated record and publishes
|
||||
** them via the configured IndexEventPublisher.
|
||||
**
|
||||
** If the publisher is null (QBit not yet initialized) or if publishing fails,
|
||||
** a warning is logged and the original records are returned without throwing,
|
||||
@@ -100,7 +86,7 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
|
||||
String primaryKeyField = (tableConfig != null && tableConfig.getPrimaryKeyField() != null) ? tableConfig.getPrimaryKeyField() : "id";
|
||||
|
||||
List<IndexEvent> events = new ArrayList<>();
|
||||
for(QRecord record : fetchCurrentRecords(updateInput, records, primaryKeyField))
|
||||
for(QRecord record : records)
|
||||
{
|
||||
events.add(new IndexEvent()
|
||||
.withAction(IndexEventAction.INDEX)
|
||||
@@ -109,10 +95,7 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
|
||||
.withRecord(record));
|
||||
}
|
||||
|
||||
if(!events.isEmpty())
|
||||
{
|
||||
publisher.publishIndexEvents(events);
|
||||
}
|
||||
publisher.publishIndexEvents(events);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
@@ -122,36 +105,4 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
|
||||
return records;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query the source table for the full, current version of each successfully
|
||||
** updated record. Uses the update's transaction, so uncommitted changes are
|
||||
** visible.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> fetchCurrentRecords(UpdateInput updateInput, List<QRecord> updatedRecords, String primaryKeyField) throws QException
|
||||
{
|
||||
List<Serializable> primaryKeys = new ArrayList<>();
|
||||
for(QRecord record : updatedRecords)
|
||||
{
|
||||
Serializable primaryKey = record.getValue(primaryKeyField);
|
||||
if(primaryKey != null && CollectionUtils.nullSafeIsEmpty(record.getErrors()))
|
||||
{
|
||||
primaryKeys.add(primaryKey);
|
||||
}
|
||||
}
|
||||
|
||||
if(primaryKeys.isEmpty())
|
||||
{
|
||||
return (new ArrayList<>());
|
||||
}
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(updateInput.getTableName());
|
||||
queryInput.setTransaction(updateInput.getTransaction());
|
||||
queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, primaryKeys)));
|
||||
|
||||
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -172,16 +172,6 @@ public class OpenSearchDocument
|
||||
*******************************************************************************/
|
||||
@JsonIgnore
|
||||
public String getDocumentId()
|
||||
{
|
||||
return (buildDocumentId(sourceTable, recordId));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Build the composite OpenSearch _id for a source table and record ID.
|
||||
*******************************************************************************/
|
||||
public static String buildDocumentId(String sourceTable, String recordId)
|
||||
{
|
||||
return (sourceTable + ":" + recordId);
|
||||
}
|
||||
|
||||
+27
-148
@@ -19,7 +19,6 @@ package com.kingsrook.qbits.quicksearch.opensearch;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@@ -33,10 +32,8 @@ import org.apache.hc.client5.http.auth.AuthScope;
|
||||
import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
|
||||
import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
|
||||
import org.apache.hc.core5.http.HttpHost;
|
||||
import org.opensearch.client.json.JsonData;
|
||||
import org.opensearch.client.json.jackson.JacksonJsonpMapper;
|
||||
import org.opensearch.client.opensearch.OpenSearchClient;
|
||||
import org.opensearch.client.opensearch._types.Conflicts;
|
||||
import org.opensearch.client.opensearch._types.FieldValue;
|
||||
import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
|
||||
import org.opensearch.client.opensearch._types.query_dsl.MultiMatchQuery;
|
||||
@@ -45,14 +42,12 @@ import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
|
||||
import org.opensearch.client.opensearch.core.BulkRequest;
|
||||
import org.opensearch.client.opensearch.core.BulkResponse;
|
||||
import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
|
||||
import org.opensearch.client.opensearch.core.DeleteByQueryResponse;
|
||||
import org.opensearch.client.opensearch.core.DeleteRequest;
|
||||
import org.opensearch.client.opensearch.core.IndexRequest;
|
||||
import org.opensearch.client.opensearch.core.SearchRequest;
|
||||
import org.opensearch.client.opensearch.core.SearchResponse;
|
||||
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
|
||||
import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
|
||||
import org.opensearch.client.opensearch.core.bulk.DeleteOperation;
|
||||
import org.opensearch.client.opensearch.core.bulk.IndexOperation;
|
||||
import org.opensearch.client.opensearch.core.search.Highlight;
|
||||
import org.opensearch.client.opensearch.core.search.HighlightField;
|
||||
@@ -109,33 +104,17 @@ public class QuickSearchOpenSearchClient implements Closeable
|
||||
String username = config.getOpensearchUsername();
|
||||
String password = config.getOpensearchPassword();
|
||||
|
||||
BasicCredentialsProvider credentialsProvider = null;
|
||||
if(username != null && password != null)
|
||||
{
|
||||
credentialsProvider = new BasicCredentialsProvider();
|
||||
BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
|
||||
credentialsProvider.setCredentials(
|
||||
new AuthScope(httpHost),
|
||||
new UsernamePasswordCredentials(username, password.toCharArray()));
|
||||
|
||||
builder.setHttpClientConfigCallback(httpClientBuilder ->
|
||||
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
|
||||
}
|
||||
|
||||
BasicCredentialsProvider finalCredentialsProvider = credentialsProvider;
|
||||
builder.setHttpClientConfigCallback(httpClientBuilder ->
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
// httpclient5 5.6+ decompresses gzip responses itself but leaves //
|
||||
// the Content-Encoding header, so the OpenSearch transport would //
|
||||
// gunzip the body a second time and fail. Let the transport alone //
|
||||
// handle compression. //
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
httpClientBuilder.disableContentCompression();
|
||||
|
||||
if(finalCredentialsProvider != null)
|
||||
{
|
||||
httpClientBuilder.setDefaultCredentialsProvider(finalCredentialsProvider);
|
||||
}
|
||||
return (httpClientBuilder);
|
||||
});
|
||||
|
||||
transport = builder.build();
|
||||
client = new OpenSearchClient(transport);
|
||||
indexName = config.getOpensearchIndexName();
|
||||
@@ -267,7 +246,28 @@ public class QuickSearchOpenSearchClient implements Closeable
|
||||
.document(doc)))));
|
||||
}
|
||||
|
||||
executeBulk(operations, result, "index");
|
||||
BulkRequest bulkRequest = BulkRequest.of(r -> r.operations(operations));
|
||||
|
||||
try
|
||||
{
|
||||
BulkResponse response = client.bulk(bulkRequest);
|
||||
|
||||
for(BulkResponseItem item : response.items())
|
||||
{
|
||||
if(item.error() != null)
|
||||
{
|
||||
result.addFailure(item.id() + ": " + item.error().reason());
|
||||
}
|
||||
else
|
||||
{
|
||||
result.addSuccess();
|
||||
}
|
||||
}
|
||||
}
|
||||
catch(IOException e)
|
||||
{
|
||||
throw new QException("Bulk index request failed: " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
return (result);
|
||||
@@ -275,85 +275,6 @@ public class QuickSearchOpenSearchClient implements Closeable
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Bulk-delete documents by their composite document IDs, in batches of at
|
||||
** most batchSize.
|
||||
**
|
||||
** Every document is attempted; one failing item does not stop the others.
|
||||
** A document that is not in the index counts as a success, because it is
|
||||
** already absent. Per-item errors are recorded via addFailure().
|
||||
**
|
||||
** @param documentIds the composite document IDs; null is treated as empty
|
||||
** @param batchSize maximum number of documents per bulk request
|
||||
** @return accumulated result across all batches
|
||||
** @throws QException if any batch request fails at the transport level
|
||||
*******************************************************************************/
|
||||
public BulkIndexResult deleteDocuments(List<String> documentIds, int batchSize) throws QException
|
||||
{
|
||||
BulkIndexResult result = new BulkIndexResult();
|
||||
|
||||
if(documentIds == null || documentIds.isEmpty())
|
||||
{
|
||||
return (result);
|
||||
}
|
||||
|
||||
int total = documentIds.size();
|
||||
|
||||
for(int start = 0; start < total; start += batchSize)
|
||||
{
|
||||
List<String> batch = documentIds.subList(start, Math.min(start + batchSize, total));
|
||||
List<BulkOperation> operations = new ArrayList<>();
|
||||
|
||||
for(String documentId : batch)
|
||||
{
|
||||
operations.add(BulkOperation.of(o -> o
|
||||
.delete(DeleteOperation.of(d -> d
|
||||
.index(indexName)
|
||||
.id(documentId)))));
|
||||
}
|
||||
|
||||
executeBulk(operations, result, "delete");
|
||||
}
|
||||
|
||||
return (result);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Send one bulk request and record each item's outcome in result.
|
||||
**
|
||||
** @param operations the bulk operations to send
|
||||
** @param result accumulator for per-item successes and failures
|
||||
** @param operationName used in the transport-failure message
|
||||
** @throws QException if the request fails at the transport level
|
||||
*******************************************************************************/
|
||||
private void executeBulk(List<BulkOperation> operations, BulkIndexResult result, String operationName) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
BulkResponse response = client.bulk(BulkRequest.of(r -> r.operations(operations)));
|
||||
|
||||
for(BulkResponseItem item : response.items())
|
||||
{
|
||||
if(item.error() != null)
|
||||
{
|
||||
result.addFailure(item.id() + ": " + item.error().reason());
|
||||
}
|
||||
else
|
||||
{
|
||||
result.addSuccess();
|
||||
}
|
||||
}
|
||||
}
|
||||
catch(IOException e)
|
||||
{
|
||||
throw new QException("Bulk " + operationName + " request failed: " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Delete a single document identified by sourceTable and recordId.
|
||||
**
|
||||
@@ -363,7 +284,7 @@ public class QuickSearchOpenSearchClient implements Closeable
|
||||
*******************************************************************************/
|
||||
public void deleteDocument(String sourceTable, String recordId) throws QException
|
||||
{
|
||||
String documentId = OpenSearchDocument.buildDocumentId(sourceTable, recordId);
|
||||
String documentId = sourceTable + ":" + recordId;
|
||||
|
||||
try
|
||||
{
|
||||
@@ -404,48 +325,6 @@ public class QuickSearchOpenSearchClient implements Closeable
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Delete a table's documents that were not (re-)indexed at or after cutoff,
|
||||
** including any that have no indexedAt value.
|
||||
**
|
||||
** Only documents visible to search are considered, so refresh the index
|
||||
** first. Documents changed while the delete runs are skipped (version
|
||||
** conflicts proceed), since they were just re-indexed.
|
||||
**
|
||||
** @param sourceTable the table whose documents to consider
|
||||
** @param cutoff documents indexed before this instant are deleted
|
||||
** @return the number of documents deleted
|
||||
** @throws QException if the OpenSearch call fails
|
||||
*******************************************************************************/
|
||||
public Long deleteDocumentsIndexedBefore(String sourceTable, Instant cutoff) throws QException
|
||||
{
|
||||
try
|
||||
{
|
||||
DeleteByQueryResponse response = client.deleteByQuery(DeleteByQueryRequest.of(r -> r
|
||||
.index(indexName)
|
||||
.conflicts(Conflicts.Proceed)
|
||||
.refresh(true)
|
||||
.query(Query.of(q -> q
|
||||
.bool(b -> b
|
||||
.filter(f -> f
|
||||
.term(t -> t
|
||||
.field("sourceTable")
|
||||
.value(FieldValue.of(sourceTable))))
|
||||
.mustNot(mn -> mn
|
||||
.range(rg -> rg
|
||||
.field("indexedAt")
|
||||
.gte(JsonData.of(cutoff.toString())))))))));
|
||||
|
||||
return (response.deleted() == null ? 0L : response.deleted());
|
||||
}
|
||||
catch(IOException e)
|
||||
{
|
||||
throw new QException("Failed to delete stale documents for table [" + sourceTable + "]: " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Execute a boosted multi-match search across the index.
|
||||
**
|
||||
|
||||
@@ -1,280 +0,0 @@
|
||||
/*
|
||||
* Copyright 2024 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.quicksearch.processes;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Process step that reconciles the search index with its source tables.
|
||||
**
|
||||
** For each table (or only the one named by the optional "tableName" input):
|
||||
** 1. Re-index every source record, paging in primary-key order so each record
|
||||
** is read exactly once, even while other rows are inserted or deleted.
|
||||
** 2. If every document indexed, refresh the index and delete the table's
|
||||
** documents that this run did not re-index (indexedAt before the run
|
||||
** started). Those have no source record: their deletes were missed.
|
||||
**
|
||||
** Unlike FullReindexStep, the table's documents are never wiped first, so
|
||||
** search keeps working during the run. When any document fails to index, the
|
||||
** stale-document delete is skipped (it would remove that record's old
|
||||
** document), and the run is marked FAILED.
|
||||
**
|
||||
** The cutoff comes from this JVM's clock, as do real-time indexedAt values on
|
||||
** this node; application nodes are assumed to keep their clocks in sync.
|
||||
*******************************************************************************/
|
||||
public class ReconcileIndexStep extends AbstractIndexingStep
|
||||
{
|
||||
private static final QLogger LOG = QLogger.getLogger(ReconcileIndexStep.class);
|
||||
|
||||
public static final String RUN_TYPE = "RECONCILE";
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Totals from reconciling one table.
|
||||
*******************************************************************************/
|
||||
private record ReconcileCounts(Integer recordsIndexed, Long documentsRemoved)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Reconcile every discovered table, or only the one named by "tableName".
|
||||
**
|
||||
** Output values: recordsIndexed (Integer) and documentsRemoved (Long),
|
||||
** summed across tables.
|
||||
**
|
||||
** @throws QException if reconciling a table fails (its run is marked FAILED)
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
|
||||
{
|
||||
String tableNameFilter = input.getValueString("tableName");
|
||||
|
||||
Integer recordsIndexed = 0;
|
||||
Long documentsRemoved = 0L;
|
||||
|
||||
for(QuickSearchableTableConfig tableConfig : getDiscoveredTables())
|
||||
{
|
||||
if(tableNameFilter != null && !tableNameFilter.isBlank() && !tableNameFilter.equals(tableConfig.getTableName()))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
ensureIndexRowExists(tableConfig.getTableName(), tableConfig);
|
||||
ReconcileCounts counts = reconcileTable(tableConfig);
|
||||
|
||||
recordsIndexed = recordsIndexed + counts.recordsIndexed();
|
||||
documentsRemoved = documentsRemoved + counts.documentsRemoved();
|
||||
}
|
||||
|
||||
output.addValue("recordsIndexed", recordsIndexed);
|
||||
output.addValue("documentsRemoved", documentsRemoved);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Reconcile one table, recording the run in quickSearchIndexRun.
|
||||
*******************************************************************************/
|
||||
private ReconcileCounts reconcileTable(QuickSearchableTableConfig tableConfig) throws QException
|
||||
{
|
||||
String tableName = tableConfig.getTableName();
|
||||
String primaryKeyField = tableConfig.getPrimaryKeyField();
|
||||
QuickSearchOpenSearchClient client = getClient();
|
||||
QuickSearchQBitConfig config = getConfig();
|
||||
|
||||
Integer indexId = queryIndexId(tableName);
|
||||
QuickSearchIndexRun run = createRunRecord(indexId, RUN_TYPE);
|
||||
|
||||
Integer totalProcessed = 0;
|
||||
Integer totalIndexed = 0;
|
||||
Integer totalErrors = 0;
|
||||
String firstError = null;
|
||||
|
||||
try
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
// every document this run writes has indexedAt at or after this //
|
||||
// instant, so afterwards any older document has no source record //
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
Instant reconcileStart = Instant.now();
|
||||
|
||||
Serializable lastPrimaryKey = null;
|
||||
while(true)
|
||||
{
|
||||
List<QRecord> batch = querySourceBatchAfter(tableName, primaryKeyField, lastPrimaryKey, config.getSourceBatchSize());
|
||||
if(batch.isEmpty())
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
List<OpenSearchDocument> documents = new ArrayList<>();
|
||||
for(QRecord record : batch)
|
||||
{
|
||||
OpenSearchDocument document = IndexingUtils.buildDocument(record, tableConfig);
|
||||
if(document != null)
|
||||
{
|
||||
documents.add(document);
|
||||
}
|
||||
}
|
||||
|
||||
if(!documents.isEmpty())
|
||||
{
|
||||
BulkIndexResult result = client.indexDocuments(documents, config.getBulkBatchSize());
|
||||
totalIndexed = totalIndexed + result.getSuccessCount();
|
||||
totalErrors = totalErrors + result.getFailureCount();
|
||||
|
||||
if(firstError == null && !result.getErrors().isEmpty())
|
||||
{
|
||||
firstError = result.getErrors().get(0);
|
||||
}
|
||||
}
|
||||
|
||||
totalProcessed = totalProcessed + batch.size();
|
||||
lastPrimaryKey = batch.get(batch.size() - 1).getValue(primaryKeyField);
|
||||
|
||||
if(lastPrimaryKey == null)
|
||||
{
|
||||
//////////////////////////////////////////////////////////////
|
||||
// paging needs a key to continue after; without one, the //
|
||||
// next query would start over and never finish //
|
||||
//////////////////////////////////////////////////////////////
|
||||
throw new QException("Source record has no value in primary key field [" + primaryKeyField + "]");
|
||||
}
|
||||
}
|
||||
|
||||
Long documentsRemoved = 0L;
|
||||
if(totalErrors.equals(0))
|
||||
{
|
||||
client.refreshIndex();
|
||||
documentsRemoved = client.deleteDocumentsIndexedBefore(tableName, reconcileStart);
|
||||
updateIndexRow(indexId, reconcileStart, totalProcessed);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG.warn("Reconcile had indexing errors; stale documents were not removed", logPair("tableName", tableName), logPair("errorCount", totalErrors));
|
||||
}
|
||||
|
||||
String finalStatus = totalErrors.equals(0) ? "COMPLETED" : "FAILED";
|
||||
completeRunRecord(run, finalStatus, totalProcessed, totalIndexed, totalErrors, firstError);
|
||||
|
||||
LOG.info("Reconcile complete", logPair("tableName", tableName), logPair("totalProcessed", totalProcessed),
|
||||
logPair("totalIndexed", totalIndexed), logPair("totalErrors", totalErrors), logPair("documentsRemoved", documentsRemoved));
|
||||
|
||||
return (new ReconcileCounts(totalIndexed, documentsRemoved));
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
LOG.warn("Reconcile failed", e, logPair("tableName", tableName));
|
||||
completeRunRecord(run, "FAILED", totalProcessed, totalIndexed, totalErrors, e.getMessage());
|
||||
throw new QException("Reconcile failed for table [" + tableName + "]: " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query the next batch of source records, in primary-key order, after the
|
||||
** given key (or from the start when it is null).
|
||||
*******************************************************************************/
|
||||
private List<QRecord> querySourceBatchAfter(String tableName, String primaryKeyField, Serializable afterPrimaryKey, Integer limit) throws QException
|
||||
{
|
||||
QQueryFilter filter = new QQueryFilter()
|
||||
.withOrderBy(new QFilterOrderBy(primaryKeyField, true))
|
||||
.withLimit(limit);
|
||||
|
||||
if(afterPrimaryKey != null)
|
||||
{
|
||||
filter.withCriteria(new QFilterCriteria(primaryKeyField, QCriteriaOperator.GREATER_THAN, afterPrimaryKey));
|
||||
}
|
||||
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(tableName);
|
||||
queryInput.setFilter(filter);
|
||||
|
||||
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Look up the ID of the quickSearchIndex row for a table.
|
||||
*******************************************************************************/
|
||||
private Integer queryIndexId(String tableName) throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(getConfig().getQuickSearchIndexTableName());
|
||||
queryInput.setFilter(new QQueryFilter()
|
||||
.withCriteria(new QFilterCriteria("tableName", QCriteriaOperator.EQUALS, tableName)));
|
||||
|
||||
List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
|
||||
if(CollectionUtils.nullSafeIsEmpty(rows))
|
||||
{
|
||||
throw new QException("No QuickSearchIndex row found for table: " + tableName);
|
||||
}
|
||||
|
||||
return (rows.get(0).getValueInteger("id"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Record a successful reconcile on the quickSearchIndex row. Records changed
|
||||
** after reconcileStart may not have been read, so basepull resumes from there.
|
||||
*******************************************************************************/
|
||||
private void updateIndexRow(Integer indexId, Instant reconcileStart, Integer recordCount) throws QException
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(getConfig().getQuickSearchIndexTableName());
|
||||
updateInput.setRecords(List.of(new QRecord()
|
||||
.withValue("id", indexId)
|
||||
.withValue("lastFullReindexTime", Instant.now())
|
||||
.withValue("lastBasepullTime", reconcileStart)
|
||||
.withValue("recordCount", recordCount)));
|
||||
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
|
||||
}
|
||||
+5
-33
@@ -24,7 +24,6 @@ import java.util.Map;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import com.kingsrook.qbits.quicksearch.processes.IndexingUtils;
|
||||
@@ -36,15 +35,13 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||
** immediately against the OpenSearch client.
|
||||
**
|
||||
** Index events are grouped by table, converted to OpenSearchDocument instances
|
||||
** using IndexingUtils, and bulk-indexed. Delete events are bulk-deleted.
|
||||
** Either way, per-document failures are raised as a QException.
|
||||
** using IndexingUtils, and bulk-indexed. Delete events are processed
|
||||
** individually.
|
||||
*******************************************************************************/
|
||||
public class SynchronousIndexEventPublisher implements IndexEventPublisher
|
||||
{
|
||||
private static final QLogger LOG = QLogger.getLogger(SynchronousIndexEventPublisher.class);
|
||||
|
||||
private static final Integer MAX_ERRORS_IN_MESSAGE = 10;
|
||||
|
||||
private final QuickSearchOpenSearchClient client;
|
||||
private final Map<String, QuickSearchableTableConfig> tableConfigsByName;
|
||||
private final Integer bulkBatchSize;
|
||||
@@ -122,16 +119,14 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
|
||||
|
||||
if(!allDocuments.isEmpty())
|
||||
{
|
||||
BulkIndexResult result = client.indexDocuments(allDocuments, bulkBatchSize);
|
||||
throwIfAnyFailed(result, "index", allDocuments.size());
|
||||
client.indexDocuments(allDocuments, bulkBatchSize);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Publish delete events synchronously as one bulk delete, so every document
|
||||
** is attempted even when some fail. Throws if any document failed.
|
||||
** Publish delete events synchronously by calling deleteDocument for each event.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void publishDeleteEvents(List<IndexEvent> events) throws QException
|
||||
@@ -141,33 +136,10 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
|
||||
return;
|
||||
}
|
||||
|
||||
List<String> documentIds = new ArrayList<>();
|
||||
for(IndexEvent event : events)
|
||||
{
|
||||
documentIds.add(OpenSearchDocument.buildDocumentId(event.getTableName(), event.getRecordId()));
|
||||
client.deleteDocument(event.getTableName(), event.getRecordId());
|
||||
}
|
||||
|
||||
BulkIndexResult result = client.deleteDocuments(documentIds, bulkBatchSize);
|
||||
throwIfAnyFailed(result, "delete", documentIds.size());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Throw a QException naming the failed documents, if any bulk item failed,
|
||||
** so failures reach the caller instead of leaving the index silently stale.
|
||||
*******************************************************************************/
|
||||
private void throwIfAnyFailed(BulkIndexResult result, String operationName, Integer attemptedCount) throws QException
|
||||
{
|
||||
if(result == null || Boolean.TRUE.equals(result.isFullySuccessful()))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
List<String> errors = result.getErrors();
|
||||
List<String> errorsShown = errors.subList(0, Math.min(errors.size(), MAX_ERRORS_IN_MESSAGE));
|
||||
throw (new QException("Failed to " + operationName + " " + result.getFailureCount() + " of " + attemptedCount
|
||||
+ " search documents: " + String.join("; ", errorsShown)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -79,7 +79,7 @@ public class BaseQuickSearchTest
|
||||
////////////////////////////////////////////////////
|
||||
// Set up fully-anonymous authentication //
|
||||
////////////////////////////////////////////////////
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
|
||||
|
||||
+2
-4
@@ -46,7 +46,6 @@ import org.junit.jupiter.api.MethodOrderer;
|
||||
import org.junit.jupiter.api.Order;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.TestMethodOrder;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
@@ -62,8 +61,7 @@ import static org.assertj.core.api.Assertions.assertThat;
|
||||
** records, runs FullReindexStep, and verifies search results including custom
|
||||
** recordLabelFormat.
|
||||
*******************************************************************************/
|
||||
@ExtendWith(RequiresDockerCondition.class)
|
||||
@Testcontainers
|
||||
@Testcontainers(disabledWithoutDocker = true)
|
||||
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
|
||||
class ConfigDrivenIndexingIntegrationTest
|
||||
{
|
||||
@@ -100,7 +98,7 @@ class ConfigDrivenIndexingIntegrationTest
|
||||
.withName(BACKEND_NAME)
|
||||
.withBackendType(MemoryBackendModule.class));
|
||||
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
|
||||
|
||||
@@ -1,283 +0,0 @@
|
||||
/*
|
||||
* Copyright 2024 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.quicksearch;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
|
||||
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
|
||||
import com.kingsrook.qbits.quicksearch.actions.QuickSearchAction;
|
||||
import com.kingsrook.qbits.quicksearch.actions.QuickSearchInput;
|
||||
import com.kingsrook.qbits.quicksearch.actions.QuickSearchResult;
|
||||
import com.kingsrook.qbits.quicksearch.annotations.QuickSearchField;
|
||||
import com.kingsrook.qbits.quicksearch.annotations.QuickSearchable;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Integration tests, against a real OpenSearch, that real-time indexing keeps
|
||||
** the index in step with the source table through updates and deletes, and
|
||||
** that the reconcile process repairs an index that has drifted.
|
||||
**
|
||||
** Each test uses its own record IDs and search terms, so tests are independent.
|
||||
*******************************************************************************/
|
||||
@ExtendWith(RequiresDockerCondition.class)
|
||||
@Testcontainers
|
||||
class IndexDriftIntegrationTest
|
||||
{
|
||||
private static final String BACKEND_NAME = "driftMemoryBackend";
|
||||
private static final String TABLE_NAME = "driftCustomer";
|
||||
private static final String INDEX_NAME = "quick_search_drift_test";
|
||||
|
||||
@Container
|
||||
static GenericContainer<?> opensearch = new GenericContainer<>("opensearchproject/opensearch:2.11.0")
|
||||
.withExposedPorts(9200)
|
||||
.withEnv("discovery.type", "single-node")
|
||||
.withEnv("plugins.security.disabled", "true")
|
||||
.withEnv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "Admin123!")
|
||||
.waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200));
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Entity annotated for quick search indexing.
|
||||
*******************************************************************************/
|
||||
@QuickSearchable(tableName = TABLE_NAME)
|
||||
static class DriftCustomer
|
||||
{
|
||||
@QuickSearchField(weight = 2)
|
||||
private String name;
|
||||
|
||||
@QuickSearchField
|
||||
private String city;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Build the instance, produce the QBit with real-time indexing on, and
|
||||
** initialize QContext.
|
||||
*******************************************************************************/
|
||||
@BeforeAll
|
||||
static void setUpAll() throws Exception
|
||||
{
|
||||
MemoryRecordStore.getInstance().reset();
|
||||
|
||||
QInstance qInstance = new QInstance();
|
||||
qInstance.addBackend(new QBackendMetaData()
|
||||
.withName(BACKEND_NAME)
|
||||
.withBackendType(MemoryBackendModule.class));
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.addTable(new QTableMetaData()
|
||||
.withName(TABLE_NAME)
|
||||
.withBackendName(BACKEND_NAME)
|
||||
.withPrimaryKeyField("id")
|
||||
.withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
|
||||
.withField(new QFieldMetaData("name", QFieldType.STRING))
|
||||
.withField(new QFieldMetaData("city", QFieldType.STRING)));
|
||||
|
||||
QuickSearchQBitConfig config = new QuickSearchQBitConfig()
|
||||
.withBackendName(BACKEND_NAME)
|
||||
.withOpensearchHost(opensearch.getHost())
|
||||
.withOpensearchPort(opensearch.getMappedPort(9200))
|
||||
.withOpensearchIndexName(INDEX_NAME)
|
||||
.withSearchableEntityClasses(List.of(DriftCustomer.class))
|
||||
.withEnableRealTimeIndexing(true);
|
||||
|
||||
new QuickSearchQBitProducer().withConfig(config).produce(qInstance);
|
||||
|
||||
QContext.init(qInstance, new QSession());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Clean up QContext and QBit state after all tests complete.
|
||||
*******************************************************************************/
|
||||
@AfterAll
|
||||
static void tearDownAll()
|
||||
{
|
||||
QContext.clear();
|
||||
QuickSearchQBitContext.clear();
|
||||
MemoryRecordStore.getInstance().reset();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a partial update (one field) keeps the other fields searchable.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPartialUpdate_unchangedFieldsStaySearchable() throws Exception
|
||||
{
|
||||
insert(new QRecord().withValue("id", 1).withValue("name", "Ada").withValue("city", "Lovelaceville"));
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue("id", 1).withValue("name", "Adeline")));
|
||||
new UpdateAction().execute(updateInput);
|
||||
|
||||
assertThat(searchRecordIds("lovelaceville"))
|
||||
.as("the unchanged city must stay searchable after updating only the name")
|
||||
.containsExactly("1");
|
||||
assertThat(searchRecordIds("adeline")).containsExactly("1");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: deleting records removes their documents, even when the batch also
|
||||
** holds a record whose document is already missing from the index.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testDelete_removesDocuments_evenWithMissingDocumentInBatch() throws Exception
|
||||
{
|
||||
insert(
|
||||
new QRecord().withValue("id", 10).withValue("name", "Tenley").withValue("city", "Deleteton"),
|
||||
new QRecord().withValue("id", 11).withValue("name", "Elvira").withValue("city", "Deleteton"),
|
||||
new QRecord().withValue("id", 12).withValue("name", "Twila").withValue("city", "Deleteton"));
|
||||
|
||||
////////////////////////////////////////////////////////////////////
|
||||
// simulate earlier drift: record 12's document is already gone //
|
||||
////////////////////////////////////////////////////////////////////
|
||||
client().deleteDocument(TABLE_NAME, "12");
|
||||
|
||||
DeleteInput deleteInput = new DeleteInput();
|
||||
deleteInput.setTableName(TABLE_NAME);
|
||||
deleteInput.setPrimaryKeys(List.<Serializable>of(12, 10));
|
||||
new DeleteAction().execute(deleteInput);
|
||||
|
||||
assertThat(searchRecordIds("deleteton"))
|
||||
.as("only the record that was not deleted should remain")
|
||||
.containsExactly("11");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the reconcile process repairs each kind of drift from the source
|
||||
** table: an orphan document (record gone), a missing document, and a stale
|
||||
** document (record changed since it was indexed).
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_repairsOrphanMissingAndStaleDocuments() throws Exception
|
||||
{
|
||||
insert(
|
||||
new QRecord().withValue("id", 21).withValue("name", "Stella").withValue("city", "Freshford"),
|
||||
new QRecord().withValue("id", 22).withValue("name", "Mona").withValue("city", "Missingham"));
|
||||
|
||||
////////////////////////////////////////////////////////////////////
|
||||
// drift the index away from the source table three ways //
|
||||
////////////////////////////////////////////////////////////////////
|
||||
Instant anHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
|
||||
client().indexDocuments(List.of(
|
||||
new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("99")
|
||||
.withSearchableText("Ghostly Phantomburg").withIndexedAt(anHourAgo).withFieldValues(Map.of()),
|
||||
new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("21")
|
||||
.withSearchableText("Stella Oldtown").withIndexedAt(anHourAgo).withFieldValues(Map.of())), 10);
|
||||
client().deleteDocument(TABLE_NAME, "22");
|
||||
|
||||
assertThat(searchRecordIds("phantomburg")).containsExactly("99");
|
||||
assertThat(searchRecordIds("oldtown")).containsExactly("21");
|
||||
assertThat(searchRecordIds("missingham")).isEmpty();
|
||||
|
||||
RunBackendStepInput input = new RunBackendStepInput();
|
||||
input.addValue("tableName", TABLE_NAME);
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new ReconcileIndexStep().run(input, output);
|
||||
|
||||
assertThat(searchRecordIds("phantomburg")).as("orphan document removed").isEmpty();
|
||||
assertThat(searchRecordIds("oldtown")).as("stale document replaced").isEmpty();
|
||||
assertThat(searchRecordIds("freshford")).as("stale document replaced").containsExactly("21");
|
||||
assertThat(searchRecordIds("missingham")).as("missing document restored").containsExactly("22");
|
||||
assertThat(output.getValue("documentsRemoved")).isEqualTo(1L);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Insert records through InsertAction (so real-time indexing runs).
|
||||
*******************************************************************************/
|
||||
private void insert(QRecord... records) throws QException
|
||||
{
|
||||
InsertInput insertInput = new InsertInput();
|
||||
insertInput.setTableName(TABLE_NAME);
|
||||
insertInput.setRecords(List.of(records));
|
||||
new InsertAction().execute(insertInput);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Refresh the index, then return the record IDs matching the term.
|
||||
*******************************************************************************/
|
||||
private List<String> searchRecordIds(String term) throws QException
|
||||
{
|
||||
client().refreshIndex();
|
||||
return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME))
|
||||
.getResults().stream()
|
||||
.map(QuickSearchResult::getRecordId)
|
||||
.toList());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The live OpenSearch client from the QBit context.
|
||||
*******************************************************************************/
|
||||
private QuickSearchOpenSearchClient client()
|
||||
{
|
||||
return (QuickSearchQBitContext.getClient());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -47,7 +47,6 @@ import org.junit.jupiter.api.MethodOrderer;
|
||||
import org.junit.jupiter.api.Order;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.TestMethodOrder;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
@@ -62,8 +61,7 @@ import static org.assertj.core.api.Assertions.assertThat;
|
||||
** Tests run in order: index creation, full reindex, search with results,
|
||||
** pagination, and no-results search.
|
||||
*******************************************************************************/
|
||||
@ExtendWith(RequiresDockerCondition.class)
|
||||
@Testcontainers
|
||||
@Testcontainers(disabledWithoutDocker = true)
|
||||
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
|
||||
class OpenSearchIntegrationTest
|
||||
{
|
||||
@@ -117,7 +115,7 @@ class OpenSearchIntegrationTest
|
||||
.withName(BACKEND_NAME)
|
||||
.withBackendType(MemoryBackendModule.class));
|
||||
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
|
||||
|
||||
@@ -162,7 +162,6 @@ class QuickSearchQBitProducerTest
|
||||
assertThat(qInstance.getTable("myApp_quickSearchIndexRun")).isNotNull();
|
||||
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
|
||||
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
|
||||
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
|
||||
}
|
||||
|
||||
|
||||
@@ -343,7 +342,6 @@ class QuickSearchQBitProducerTest
|
||||
|
||||
assertThat(qInstance.getProcess(QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
|
||||
assertThat(qInstance.getProcess(QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
|
||||
assertThat(qInstance.getProcess(QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
|
||||
}
|
||||
|
||||
|
||||
@@ -796,7 +794,7 @@ class QuickSearchQBitProducerTest
|
||||
.withName(BACKEND_NAME)
|
||||
.withBackendType(MemoryBackendModule.class));
|
||||
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
|
||||
|
||||
@@ -1,72 +0,0 @@
|
||||
/*
|
||||
* Copyright 2024 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.quicksearch;
|
||||
|
||||
|
||||
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
|
||||
import org.junit.jupiter.api.extension.ExecutionCondition;
|
||||
import org.junit.jupiter.api.extension.ExtensionContext;
|
||||
import org.testcontainers.DockerClientFactory;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** JUnit condition for integration tests that need Docker.
|
||||
**
|
||||
** Without Docker, the tests are skipped on a developer machine, but fail when
|
||||
** CI=true, so a CI build can never pass with its integration tests silently
|
||||
** skipped. Register it ahead of @Testcontainers, so no container is started
|
||||
** when the tests are disabled.
|
||||
*******************************************************************************/
|
||||
public class RequiresDockerCondition implements ExecutionCondition
|
||||
{
|
||||
|
||||
/*******************************************************************************
|
||||
** Evaluate against the local Docker environment and the CI env var.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context)
|
||||
{
|
||||
return (evaluate(DockerClientFactory.instance().isDockerAvailable(), System.getenv("CI")));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Decide whether Docker-backed tests run, skip, or fail.
|
||||
**
|
||||
** @param dockerAvailable whether a usable Docker environment was found
|
||||
** @param ciValue value of the CI environment variable (may be null)
|
||||
** @return enabled when Docker is available; disabled when it is not and CI
|
||||
** is not "true"
|
||||
** @throws IllegalStateException when Docker is unavailable and CI is "true"
|
||||
*******************************************************************************/
|
||||
static ConditionEvaluationResult evaluate(Boolean dockerAvailable, String ciValue)
|
||||
{
|
||||
if(Boolean.TRUE.equals(dockerAvailable))
|
||||
{
|
||||
return (ConditionEvaluationResult.enabled("Docker is available"));
|
||||
}
|
||||
|
||||
if("true".equalsIgnoreCase(ciValue))
|
||||
{
|
||||
throw (new IllegalStateException("Docker is required for integration tests when CI=true, but no usable Docker environment was found"));
|
||||
}
|
||||
|
||||
return (ConditionEvaluationResult.disabled("Docker is not available; skipping (with CI=true these tests fail instead)"));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,75 +0,0 @@
|
||||
/*
|
||||
* Copyright 2024 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.quicksearch;
|
||||
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Tests for RequiresDockerCondition.
|
||||
*******************************************************************************/
|
||||
class RequiresDockerConditionTest
|
||||
{
|
||||
|
||||
/*******************************************************************************
|
||||
** Docker available: integration tests run, whether or not CI is set.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testEvaluate_dockerAvailable_enabled()
|
||||
{
|
||||
ConditionEvaluationResult local = RequiresDockerCondition.evaluate(true, null);
|
||||
ConditionEvaluationResult ci = RequiresDockerCondition.evaluate(true, "true");
|
||||
|
||||
assertThat(local.isDisabled()).isFalse();
|
||||
assertThat(ci.isDisabled()).isFalse();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** No Docker outside CI: integration tests are skipped.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testEvaluate_noDockerOutsideCi_disabled()
|
||||
{
|
||||
assertThat(RequiresDockerCondition.evaluate(false, null).isDisabled()).isTrue();
|
||||
assertThat(RequiresDockerCondition.evaluate(false, "").isDisabled()).isTrue();
|
||||
assertThat(RequiresDockerCondition.evaluate(false, "false").isDisabled()).isTrue();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** No Docker with CI=true: evaluation throws, so the tests fail instead of
|
||||
** silently skipping.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testEvaluate_noDockerInCi_throws()
|
||||
{
|
||||
assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "true"))
|
||||
.isInstanceOf(IllegalStateException.class)
|
||||
.hasMessageContaining("CI=true");
|
||||
|
||||
assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "TRUE"))
|
||||
.isInstanceOf(IllegalStateException.class);
|
||||
}
|
||||
|
||||
}
|
||||
+1
-46
@@ -29,7 +29,6 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
|
||||
@@ -80,7 +79,7 @@ class QuickSearchPostDeleteCustomizerTest
|
||||
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
|
||||
.withField(new QFieldMetaData("name", QFieldType.STRING)));
|
||||
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
|
||||
QContext.init(qInstance, new QSession());
|
||||
}
|
||||
@@ -145,50 +144,6 @@ class QuickSearchPostDeleteCustomizerTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that records the delete rejected (with errors), or that carry no
|
||||
** primary key, do not remove documents from the index.
|
||||
*******************************************************************************/
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
void testPostDelete_recordsWithErrorsOrNoPrimaryKey_skipped() throws QException
|
||||
{
|
||||
QRecord deleted = new QRecord().withValue(PK_FIELD, 5);
|
||||
QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
|
||||
rejected.addError(new BadInputStatusMessage("still referenced"));
|
||||
QRecord noKey = new QRecord().withValue("name", "orphan");
|
||||
|
||||
DeleteInput deleteInput = new DeleteInput();
|
||||
deleteInput.setTableName(TABLE_NAME);
|
||||
|
||||
List<QRecord> result = new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(deleted, rejected, noKey));
|
||||
assertThat(result).containsExactly(deleted, rejected, noKey);
|
||||
|
||||
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(mockPublisher).publishDeleteEvents(captor.capture());
|
||||
assertThat(captor.getValue()).extracting(IndexEvent::getRecordId).containsExactly("5");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that when no record is eligible, nothing is published.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPostDelete_onlyRejectedRecords_nothingPublished() throws QException
|
||||
{
|
||||
QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
|
||||
rejected.addError(new BadInputStatusMessage("still referenced"));
|
||||
|
||||
DeleteInput deleteInput = new DeleteInput();
|
||||
deleteInput.setTableName(TABLE_NAME);
|
||||
|
||||
new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(rejected));
|
||||
verify(mockPublisher, never()).publishDeleteEvents(anyList());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that post-delete with empty records list returns input without calling
|
||||
** publisher.
|
||||
|
||||
+1
-1
@@ -79,7 +79,7 @@ class QuickSearchPostInsertCustomizerTest
|
||||
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
|
||||
.withField(new QFieldMetaData("name", QFieldType.STRING)));
|
||||
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
|
||||
QContext.init(qInstance, new QSession());
|
||||
}
|
||||
|
||||
+29
-149
@@ -19,26 +19,17 @@ package com.kingsrook.qbits.quicksearch.customizers;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.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.QBackendMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
|
||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
|
||||
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
|
||||
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
|
||||
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
|
||||
@@ -59,29 +50,23 @@ import static org.mockito.Mockito.verify;
|
||||
|
||||
/*******************************************************************************
|
||||
** Tests for QuickSearchPostUpdateCustomizer.
|
||||
**
|
||||
** Uses a memory backend, because the customizer re-reads updated records from
|
||||
** the source table before indexing them.
|
||||
*******************************************************************************/
|
||||
class QuickSearchPostUpdateCustomizerTest
|
||||
{
|
||||
private static final String BACKEND_NAME = "memoryBackend";
|
||||
private static final String TABLE_NAME = "orders";
|
||||
private static final String PK_FIELD = "id";
|
||||
private static final String TABLE_NAME = "orders";
|
||||
private static final String PK_FIELD = "id";
|
||||
|
||||
private IndexEventPublisher mockPublisher;
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Set up mock publisher in QuickSearchQBitContext and a QContext with a
|
||||
** memory-backed test QInstance before each test.
|
||||
** Set up mock publisher in QuickSearchQBitContext and a QContext with a test
|
||||
** QInstance before each test.
|
||||
*******************************************************************************/
|
||||
@BeforeEach
|
||||
void setUp()
|
||||
{
|
||||
MemoryRecordStore.getInstance().reset();
|
||||
|
||||
mockPublisher = mock(IndexEventPublisher.class);
|
||||
QuickSearchQBitContext.setPublisher(mockPublisher);
|
||||
QuickSearchQBitContext.setDiscoveredTables(List.of(
|
||||
@@ -89,19 +74,14 @@ class QuickSearchPostUpdateCustomizerTest
|
||||
));
|
||||
|
||||
QInstance qInstance = new QInstance();
|
||||
qInstance.addBackend(new QBackendMetaData()
|
||||
.withName(BACKEND_NAME)
|
||||
.withBackendType(MemoryBackendModule.class));
|
||||
qInstance.addTable(new QTableMetaData()
|
||||
.withName(TABLE_NAME)
|
||||
.withBackendName(BACKEND_NAME)
|
||||
.withPrimaryKeyField(PK_FIELD)
|
||||
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
|
||||
.withField(new QFieldMetaData("name", QFieldType.STRING))
|
||||
.withField(new QFieldMetaData("description", QFieldType.STRING)));
|
||||
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
|
||||
.withName("anonymous")
|
||||
.withType(QAuthenticationType.FULLY_ANONYMOUS));
|
||||
.withField(new QFieldMetaData("name", QFieldType.STRING)));
|
||||
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
|
||||
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
|
||||
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
|
||||
QContext.init(qInstance, new QSession());
|
||||
}
|
||||
|
||||
@@ -115,33 +95,6 @@ class QuickSearchPostUpdateCustomizerTest
|
||||
{
|
||||
QuickSearchQBitContext.clear();
|
||||
QContext.clear();
|
||||
MemoryRecordStore.getInstance().reset();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Insert full records into the source table.
|
||||
*******************************************************************************/
|
||||
private void insertSourceRecords(QRecord... records) throws QException
|
||||
{
|
||||
InsertInput insertInput = new InsertInput();
|
||||
insertInput.setTableName(TABLE_NAME);
|
||||
insertInput.setRecords(List.of(records));
|
||||
new InsertAction().execute(insertInput);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Capture the events passed to the publisher's publishIndexEvents.
|
||||
*******************************************************************************/
|
||||
@SuppressWarnings("unchecked")
|
||||
private List<IndexEvent> captureIndexEvents() throws QException
|
||||
{
|
||||
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(mockPublisher).publishIndexEvents(captor.capture());
|
||||
return (captor.getValue());
|
||||
}
|
||||
|
||||
|
||||
@@ -158,107 +111,36 @@ class QuickSearchPostUpdateCustomizerTest
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that post-update builds INDEX events from the stored source records,
|
||||
** not from the (possibly sparse) records passed in by the update.
|
||||
** Test that post-update with valid records builds correct INDEX events and
|
||||
** calls publisher.publishIndexEvents.
|
||||
*******************************************************************************/
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
void testPostUpdate_withRecords_indexesStoredSourceRecords() throws QException
|
||||
void testPostUpdate_withRecords_publishesIndexEvents() throws QException
|
||||
{
|
||||
insertSourceRecords(
|
||||
new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated").withValue("description", "Engineer"),
|
||||
new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated").withValue("description", "Designer"));
|
||||
|
||||
QRecord sparse1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
|
||||
QRecord sparse2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
|
||||
QRecord record1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
|
||||
QRecord record2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(TABLE_NAME);
|
||||
|
||||
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
|
||||
List<QRecord> result = customizer.postUpdate(updateInput, List.of(sparse1, sparse2), Optional.empty());
|
||||
List<QRecord> result = customizer.postUpdate(updateInput, List.of(record1, record2), Optional.empty());
|
||||
|
||||
assertThat(result).containsExactly(sparse1, sparse2);
|
||||
assertThat(result).containsExactly(record1, record2);
|
||||
|
||||
List<IndexEvent> events = captureIndexEvents();
|
||||
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(mockPublisher).publishIndexEvents(captor.capture());
|
||||
|
||||
List<IndexEvent> events = captor.getValue();
|
||||
assertThat(events).hasSize(2);
|
||||
assertThat(events).allSatisfy(e ->
|
||||
{
|
||||
assertThat(e.getAction()).isEqualTo(IndexEventAction.INDEX);
|
||||
assertThat(e.getTableName()).isEqualTo(TABLE_NAME);
|
||||
});
|
||||
assertThat(events).extracting(IndexEvent::getRecordId).containsExactlyInAnyOrder("10", "20");
|
||||
|
||||
IndexEvent aliceEvent = events.stream().filter(e -> "10".equals(e.getRecordId())).findFirst().orElseThrow();
|
||||
assertThat(aliceEvent.getRecord().getValueString("name")).isEqualTo("Alice Updated");
|
||||
assertThat(aliceEvent.getRecord().getValueString("description")).isEqualTo("Engineer");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** End to end through UpdateAction: a partial update of one field must not drop
|
||||
** the other indexed fields from the document.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testUpdateAction_partialUpdate_reindexesFullRecord() throws QException
|
||||
{
|
||||
QContext.getQInstance().getTable(TABLE_NAME)
|
||||
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(QuickSearchPostUpdateCustomizer.class));
|
||||
|
||||
insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice").withValue("description", "Engineer"));
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(TABLE_NAME);
|
||||
updateInput.setRecords(List.of(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alicia")));
|
||||
new UpdateAction().execute(updateInput);
|
||||
|
||||
List<IndexEvent> events = captureIndexEvents();
|
||||
assertThat(events).hasSize(1);
|
||||
assertThat(events.get(0).getRecordId()).isEqualTo("1");
|
||||
assertThat(events.get(0).getRecord().getValueString("name")).isEqualTo("Alicia");
|
||||
assertThat(events.get(0).getRecord().getValueString("description")).isEqualTo("Engineer");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that records the update rejected (with errors) are not re-indexed.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPostUpdate_recordWithErrors_notIndexed() throws QException
|
||||
{
|
||||
insertSourceRecords(
|
||||
new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good"),
|
||||
new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected"));
|
||||
|
||||
QRecord good = new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good");
|
||||
QRecord rejected = new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected");
|
||||
rejected.addError(new BadInputStatusMessage("not allowed"));
|
||||
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(TABLE_NAME);
|
||||
|
||||
new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(good, rejected), Optional.empty());
|
||||
|
||||
List<IndexEvent> events = captureIndexEvents();
|
||||
assertThat(events).extracting(IndexEvent::getRecordId).containsExactly("1");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test that when no updated record can be read back from the source table,
|
||||
** nothing is published.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPostUpdate_recordNotInSource_nothingPublished() throws QException
|
||||
{
|
||||
UpdateInput updateInput = new UpdateInput();
|
||||
updateInput.setTableName(TABLE_NAME);
|
||||
|
||||
new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(new QRecord().withValue(PK_FIELD, 99)), Optional.empty());
|
||||
|
||||
verify(mockPublisher, never()).publishIndexEvents(anyList());
|
||||
assertThat(events.get(0).getAction()).isEqualTo(IndexEventAction.INDEX);
|
||||
assertThat(events.get(0).getTableName()).isEqualTo(TABLE_NAME);
|
||||
assertThat(events.get(0).getRecordId()).isEqualTo("10");
|
||||
assertThat(events.get(0).getRecord()).isSameAs(record1);
|
||||
assertThat(events.get(1).getAction()).isEqualTo(IndexEventAction.INDEX);
|
||||
assertThat(events.get(1).getRecordId()).isEqualTo("20");
|
||||
assertThat(events.get(1).getRecord()).isSameAs(record2);
|
||||
}
|
||||
|
||||
|
||||
@@ -327,7 +209,6 @@ class QuickSearchPostUpdateCustomizerTest
|
||||
@Test
|
||||
void testPostUpdate_publisherThrows_returnsRecordsWithoutThrowing() throws QException
|
||||
{
|
||||
insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice"));
|
||||
doThrow(new QException("publish failed"))
|
||||
.when(mockPublisher).publishIndexEvents(anyList());
|
||||
|
||||
@@ -337,7 +218,6 @@ class QuickSearchPostUpdateCustomizerTest
|
||||
|
||||
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
|
||||
assertThatNoException().isThrownBy(() -> customizer.postUpdate(updateInput, List.of(record), Optional.empty()));
|
||||
verify(mockPublisher).publishIndexEvents(anyList());
|
||||
}
|
||||
|
||||
}
|
||||
-47
@@ -261,51 +261,4 @@ class QuickSearchOpenSearchClientTest
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** deleteDocuments with a null or empty list is a no-op, even with no server.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testDeleteDocuments_emptyList_returnsEmptyResult() throws QException
|
||||
{
|
||||
QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
|
||||
|
||||
try
|
||||
{
|
||||
BulkIndexResult emptyResult = client.deleteDocuments(List.of(), 100);
|
||||
BulkIndexResult nullResult = client.deleteDocuments(null, 100);
|
||||
|
||||
assertThat(emptyResult.getSuccessCount()).isEqualTo(0);
|
||||
assertThat(emptyResult.getFailureCount()).isEqualTo(0);
|
||||
assertThat(nullResult.getSuccessCount()).isEqualTo(0);
|
||||
}
|
||||
finally
|
||||
{
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** deleteDocuments throws QException when the server is unavailable.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testDeleteDocuments_serverUnavailable_throwsQException() throws QException
|
||||
{
|
||||
QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
|
||||
|
||||
try
|
||||
{
|
||||
assertThatThrownBy(() -> client.deleteDocuments(List.of("orders:1", "orders:2"), 100))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("Bulk delete");
|
||||
}
|
||||
finally
|
||||
{
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,281 +0,0 @@
|
||||
/*
|
||||
* Copyright 2024 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.quicksearch.processes;
|
||||
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
|
||||
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
|
||||
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
|
||||
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InOrder;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyInt;
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.atLeastOnce;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Tests for ReconcileIndexStep.
|
||||
*******************************************************************************/
|
||||
class ReconcileIndexStepTest extends BaseQuickSearchTest
|
||||
{
|
||||
private QuickSearchOpenSearchClient mockClient;
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Install a mock client that reports every document as indexed.
|
||||
*******************************************************************************/
|
||||
@BeforeEach
|
||||
void setUpClient() throws QException
|
||||
{
|
||||
mockClient = mock(QuickSearchOpenSearchClient.class);
|
||||
when(mockClient.indexDocuments(anyList(), anyInt())).thenAnswer(invocation ->
|
||||
{
|
||||
List<?> docs = invocation.getArgument(0);
|
||||
return (new BulkIndexResult().withSuccessCount(docs.size()));
|
||||
});
|
||||
when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(0L);
|
||||
QuickSearchQBitContext.setClient(mockClient);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Insert count source records into the testEntity table.
|
||||
*******************************************************************************/
|
||||
private void insertTestEntities(int count) throws QException
|
||||
{
|
||||
List<QRecord> records = new ArrayList<>();
|
||||
for(int i = 1; i <= count; i++)
|
||||
{
|
||||
records.add(new QRecord().withValue("name", "Entity " + i).withValue("description", "Description " + i));
|
||||
}
|
||||
|
||||
InsertInput insertInput = new InsertInput();
|
||||
insertInput.setTableName(TEST_ENTITY_TABLE);
|
||||
insertInput.setRecords(records);
|
||||
new InsertAction().execute(insertInput);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query all rows of the given table.
|
||||
*******************************************************************************/
|
||||
private List<QRecord> queryAll(String tableName) throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput();
|
||||
queryInput.setTableName(tableName);
|
||||
queryInput.setFilter(new QQueryFilter());
|
||||
return (new QueryAction().execute(queryInput).getRecords());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Collect the record IDs of every document passed to indexDocuments.
|
||||
*******************************************************************************/
|
||||
@SuppressWarnings("unchecked")
|
||||
private List<String> indexedRecordIds() throws QException
|
||||
{
|
||||
ArgumentCaptor<List<OpenSearchDocument>> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(mockClient, atLeastOnce()).indexDocuments(captor.capture(), anyInt());
|
||||
|
||||
List<String> recordIds = new ArrayList<>();
|
||||
for(List<OpenSearchDocument> batch : captor.getAllValues())
|
||||
{
|
||||
batch.forEach(doc -> recordIds.add(doc.getRecordId()));
|
||||
}
|
||||
return (recordIds);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: every source record is re-indexed, then documents not re-indexed by
|
||||
** this run are removed, after a refresh. The table is never wiped first.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_reindexesSourceThenRemovesStaleDocuments() throws QException
|
||||
{
|
||||
insertTestEntities(3);
|
||||
when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(2L);
|
||||
|
||||
Instant beforeRun = Instant.now();
|
||||
RunBackendStepOutput output = new RunBackendStepOutput();
|
||||
new ReconcileIndexStep().run(new RunBackendStepInput(), output);
|
||||
Instant afterRun = Instant.now();
|
||||
|
||||
assertThat(indexedRecordIds()).containsExactlyInAnyOrder("1", "2", "3");
|
||||
|
||||
ArgumentCaptor<Instant> cutoffCaptor = ArgumentCaptor.forClass(Instant.class);
|
||||
InOrder inOrder = inOrder(mockClient);
|
||||
inOrder.verify(mockClient).indexDocuments(anyList(), anyInt());
|
||||
inOrder.verify(mockClient).refreshIndex();
|
||||
inOrder.verify(mockClient).deleteDocumentsIndexedBefore(eq(TEST_ENTITY_TABLE), cutoffCaptor.capture());
|
||||
|
||||
assertThat(cutoffCaptor.getValue()).isBetween(beforeRun, afterRun);
|
||||
verify(mockClient, never()).deleteDocumentsForTable(anyString());
|
||||
|
||||
assertThat(output.getValueInteger("recordsIndexed")).isEqualTo(3);
|
||||
assertThat(output.getValue("documentsRemoved")).isEqualTo(2L);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: source records are paged by primary key, so every record is read
|
||||
** exactly once across batches.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_pagesByPrimaryKey_eachRecordOnce() throws QException
|
||||
{
|
||||
QuickSearchQBitContext.getConfig().withSourceBatchSize(2);
|
||||
insertTestEntities(5);
|
||||
|
||||
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
verify(mockClient, times(3)).indexDocuments(anyList(), anyInt());
|
||||
assertThat(indexedRecordIds()).containsExactly("1", "2", "3", "4", "5");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: when any document fails to index, stale documents are not removed
|
||||
** (a document that failed to re-index would otherwise be purged), and the
|
||||
** run is marked FAILED.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_indexingErrors_keepsDocumentsAndFailsRun() throws QException
|
||||
{
|
||||
insertTestEntities(2);
|
||||
BulkIndexResult partial = new BulkIndexResult();
|
||||
partial.addSuccess();
|
||||
partial.addFailure("testEntity:2: mapper_parsing_exception");
|
||||
when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(partial);
|
||||
|
||||
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
|
||||
|
||||
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
|
||||
assertThat(runs).hasSize(1);
|
||||
assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
|
||||
assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
|
||||
assertThat(runs.get(0).getValueInteger("errorCount")).isEqualTo(1);
|
||||
assertThat(runs.get(0).getValueString("errorMessage")).contains("mapper_parsing_exception");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a client exception fails the run record and is rethrown.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_clientThrows_failsRunAndThrows() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
when(mockClient.indexDocuments(anyList(), anyInt())).thenThrow(new QException("connection refused"));
|
||||
|
||||
assertThatThrownBy(() -> new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("Reconcile failed");
|
||||
|
||||
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
|
||||
assertThat(runs).hasSize(1);
|
||||
assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
|
||||
assertThat(runs.get(0).getValueString("errorMessage")).contains("connection refused");
|
||||
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: a successful run completes its run record and updates the index row.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_success_completesRunAndUpdatesIndexRow() throws QException
|
||||
{
|
||||
insertTestEntities(4);
|
||||
|
||||
Instant beforeRun = Instant.now();
|
||||
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
|
||||
|
||||
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
|
||||
assertThat(runs).hasSize(1);
|
||||
assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
|
||||
assertThat(runs.get(0).getValueString("status")).isEqualTo("COMPLETED");
|
||||
assertThat(runs.get(0).getValueInteger("recordsProcessed")).isEqualTo(4);
|
||||
assertThat(runs.get(0).getValueInteger("recordsIndexed")).isEqualTo(4);
|
||||
|
||||
List<QRecord> indexRows = queryAll(QuickSearchIndex.TABLE_NAME);
|
||||
assertThat(indexRows).hasSize(1);
|
||||
assertThat(indexRows.get(0).getValueInteger("recordCount")).isEqualTo(4);
|
||||
assertThat(indexRows.get(0).getValueInstant("lastFullReindexTime")).isAfterOrEqualTo(beforeRun);
|
||||
assertThat(indexRows.get(0).getValueInstant("lastBasepullTime")).isAfterOrEqualTo(beforeRun);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test: the tableName input limits the run to that table.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testReconcile_unknownTableNameInput_nothingReconciled() throws QException
|
||||
{
|
||||
insertTestEntities(1);
|
||||
|
||||
RunBackendStepInput input = new RunBackendStepInput();
|
||||
input.addValue("tableName", "someOtherTable");
|
||||
new ReconcileIndexStep().run(input, new RunBackendStepOutput());
|
||||
|
||||
verify(mockClient, never()).indexDocuments(anyList(), anyInt());
|
||||
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
|
||||
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).isEmpty();
|
||||
}
|
||||
|
||||
}
|
||||
+5
-58
@@ -34,7 +34,6 @@ import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.anyInt;
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
@@ -168,11 +167,10 @@ class SynchronousIndexEventPublisherTest
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test publishDeleteEvents removes all documents in one bulk call, so one
|
||||
** failing document cannot stop the others from being removed.
|
||||
** Test publishDeleteEvents calls client.deleteDocument for each event.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPublishDeleteEvents_bulkDeletesAllDocuments() throws QException
|
||||
void testPublishDeleteEvents_callsClientForEachEvent() throws QException
|
||||
{
|
||||
List<IndexEvent> events = List.of(
|
||||
new IndexEvent()
|
||||
@@ -184,61 +182,10 @@ class SynchronousIndexEventPublisherTest
|
||||
.withRecordId("20")
|
||||
.withAction(IndexEventAction.DELETE));
|
||||
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult());
|
||||
|
||||
publisher.publishDeleteEvents(events);
|
||||
|
||||
verify(mockClient).deleteDocuments(List.of("orders:10", "orders:20"), BATCH_SIZE);
|
||||
verify(mockClient, never()).deleteDocument(anyString(), anyString());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test publishDeleteEvents throws, naming the failed documents, when some
|
||||
** bulk items fail.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPublishDeleteEvents_itemFailures_throwsQException() throws QException
|
||||
{
|
||||
BulkIndexResult result = new BulkIndexResult();
|
||||
result.addSuccess();
|
||||
result.addFailure("orders:20: shard unavailable");
|
||||
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(result);
|
||||
|
||||
List<IndexEvent> events = List.of(
|
||||
new IndexEvent().withTableName("orders").withRecordId("10").withAction(IndexEventAction.DELETE),
|
||||
new IndexEvent().withTableName("orders").withRecordId("20").withAction(IndexEventAction.DELETE));
|
||||
|
||||
assertThatThrownBy(() -> publisher.publishDeleteEvents(events))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("1 of 2")
|
||||
.hasMessageContaining("orders:20: shard unavailable");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Test publishIndexEvents throws, naming the failed documents, when some
|
||||
** bulk items fail (instead of silently dropping the failures).
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testPublishIndexEvents_itemFailures_throwsQException() throws QException
|
||||
{
|
||||
BulkIndexResult result = new BulkIndexResult();
|
||||
result.addFailure("orders:1: mapper_parsing_exception");
|
||||
when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(result);
|
||||
|
||||
IndexEvent event = new IndexEvent()
|
||||
.withTableName("orders")
|
||||
.withRecordId("1")
|
||||
.withAction(IndexEventAction.INDEX)
|
||||
.withRecord(new QRecord().withValue("id", 1).withValue("orderNumber", "ORD-001"));
|
||||
|
||||
assertThatThrownBy(() -> publisher.publishIndexEvents(List.of(event)))
|
||||
.isInstanceOf(QException.class)
|
||||
.hasMessageContaining("1 of 1")
|
||||
.hasMessageContaining("orders:1: mapper_parsing_exception");
|
||||
verify(mockClient).deleteDocument("orders", "10");
|
||||
verify(mockClient).deleteDocument("orders", "20");
|
||||
}
|
||||
|
||||
|
||||
@@ -262,7 +209,7 @@ class SynchronousIndexEventPublisherTest
|
||||
void testPublishDeleteEvents_clientThrows_propagatesException() throws QException
|
||||
{
|
||||
doThrow(new QException("delete failed"))
|
||||
.when(mockClient).deleteDocuments(anyList(), anyInt());
|
||||
.when(mockClient).deleteDocument("orders", "1");
|
||||
|
||||
IndexEvent event = new IndexEvent()
|
||||
.withTableName("orders")
|
||||
|
||||
Reference in new issue
Block a user