From f2703f481800f63b883f3b09d2e700dd12149e47 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 6 Oct 2026 18:21:51 -0500 Subject: [PATCH 1/2] fix(#13): record failed events when the listener re-read fails A failed in-transaction re-read of updated records now records their keys in quickSearchFailedEvent after the commit, so basepull replays them. A LinkageError from building or invoking the lazy client and publisher is recorded the same way, and a DEGRADED start tolerates it. Closes #13 --- .../qbits/quicksearch/QuickSearchRuntime.java | 6 +- .../QuickSearchRecordChangeListener.java | 58 +++++-- .../MissingLibraryTransportCustomizer.java | 43 +++++ .../quicksearch/QuickSearchRuntimeTest.java | 18 +++ .../QuickSearchRecordChangeListenerTest.java | 152 ++++++++++++++++++ 5 files changed, 261 insertions(+), 16 deletions(-) create mode 100644 src/test/java/com/kingsrook/qbits/quicksearch/MissingLibraryTransportCustomizer.java diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java index 94c22da..dd5dd35 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java @@ -338,7 +338,9 @@ public class QuickSearchRuntime /******************************************************************************* ** Start the runtime per the configured startup mode: FAIL_FAST throws when - ** the index cannot be prepared; DEGRADED logs and continues. + ** the index cannot be prepared; DEGRADED logs and continues. A LinkageError + ** (an optional library such as the AWS SDK missing while the client is + ** built) is handled the same way, so a DEGRADED host still boots. *******************************************************************************/ public void start() throws QException { @@ -346,7 +348,7 @@ public class QuickSearchRuntime { ensureIndexReady(); } - catch(Exception e) + catch(Exception | LinkageError e) { if(config.getStartupMode() == QuickSearchStartupMode.DEGRADED) { diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java index 990f884..c35a57c 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java @@ -57,9 +57,12 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; ** is already committed, so it publishes immediately. A rolled-back write never ** reaches the index. ** - ** This listener never fails the host's write. When publishing fails, each - ** affected (table, record, action) is written to quickSearchFailedEvent for - ** the scheduled basepull to replay. + ** This listener never fails the host's write. When the re-read of updated + ** records or publishing fails (including a LinkageError from a missing + ** optional library while the client is built), each affected (table, record, + ** action) is written to quickSearchFailedEvent for the scheduled basepull to + ** replay. A failed re-read is recorded after the commit too, so a rolled-back + ** write leaves no row. ** ** While a full reindex rebuilds into a new physical index, the alias still ** points at the old one, so a published delete never reaches the new index. @@ -113,7 +116,22 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte switch(event.getType()) { case INSERT -> indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, event.getRecords())); - case UPDATE -> indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, fetchCurrentRecords(event, primaryKeyField))); + case UPDATE -> + { + try + { + indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, fetchCurrentRecords(event, primaryKeyField))); + } + catch(Exception e) + { + //////////////////////////////////////////////////////////////////// + // record the updated keys, so basepull re-reads and indexes them // + //////////////////////////////////////////////////////////////////// + List failedEvents = toIndexEvents(tableName, primaryKeyField, event.getRecords()); + runAfterCommit(event, () -> recordFailures(runtime, failedEvents, e)); + return; + } + } case DELETE -> deleteEvents.addAll(toDeleteEvents(tableName, primaryKeyField, event.getRecords())); } @@ -122,15 +140,24 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte return; } - Runnable publish = () -> publish(runtime, indexEvents, deleteEvents); + runAfterCommit(event, () -> publish(runtime, indexEvents, deleteEvents)); + } + + + /******************************************************************************* + ** Run work once the event's transaction commits, or now when the write had + ** no transaction (it is already committed). + *******************************************************************************/ + private static void runAfterCommit(RecordChangeEvent event, Runnable work) + { if(event.getTransaction() != null) { - event.getTransaction().addAfterCommitCallback(publish); + event.getTransaction().addAfterCommitCallback(work); } else { - publish.run(); + work.run(); } } @@ -161,7 +188,10 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* - ** + ** A LinkageError is caught along with Exception: building the lazy client, + ** or a host publisher, can hit an optional library that is not on the + ** classpath (the AWS SDK for AWS_SIGV4), which a DEGRADED start leaves for + ** the first write to find. *******************************************************************************/ private static void publishOrRecordFailures(QuickSearchRuntime runtime, List indexEvents, List deleteEvents) { @@ -170,7 +200,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { publisher = runtime.getPublisher(); } - catch(Exception e) + catch(Exception | LinkageError e) { recordFailures(runtime, indexEvents, e); recordFailures(runtime, deleteEvents, e); @@ -183,7 +213,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { publisher.publishIndexEvents(indexEvents); } - catch(Exception e) + catch(Exception | LinkageError e) { recordFailures(runtime, indexEvents, e); } @@ -195,7 +225,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { publisher.publishDeleteEvents(deleteEvents); } - catch(Exception e) + catch(Exception | LinkageError e) { recordFailures(runtime, deleteEvents, e); } @@ -205,9 +235,9 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* - ** Write one quickSearchFailedEvent row per event. + ** Write one quickSearchFailedEvent row per event. Never throws. *******************************************************************************/ - static void recordFailures(QuickSearchRuntime runtime, List events, Exception cause) + static void recordFailures(QuickSearchRuntime runtime, List events, Throwable cause) { if(events.isEmpty()) { @@ -215,7 +245,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte } String message = cause.getMessage() == null ? cause.getClass().getSimpleName() : cause.getMessage(); - LOG.warn("Real-time index publish failed; recording failed events for replay", cause, + LOG.warn("Real-time indexing failed; recording failed events for replay", cause, logPair("tableName", events.get(0).getTableName()), logPair("eventCount", events.size())); try diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/MissingLibraryTransportCustomizer.java b/src/test/java/com/kingsrook/qbits/quicksearch/MissingLibraryTransportCustomizer.java new file mode 100644 index 0000000..994840b --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/quicksearch/MissingLibraryTransportCustomizer.java @@ -0,0 +1,43 @@ +/* + * 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 com.kingsrook.qbits.quicksearch.opensearch.OpenSearchTransportCustomizer; +import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder; + + +/******************************************************************************* + ** Test transport customizer that fails linkage while the OpenSearch client is + ** built, as a missing optional library (the AWS SDK for AWS_SIGV4) does. + *******************************************************************************/ +public class MissingLibraryTransportCustomizer implements OpenSearchTransportCustomizer +{ + public static final String MISSING_CLASS = "software/amazon/awssdk/http/SdkHttpClient"; + + + + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public ApacheHttpClient5TransportBuilder customizeTransport(ApacheHttpClient5TransportBuilder builder) + { + throw (new NoClassDefFoundError(MISSING_CLASS)); + } + +} diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java index 8beccac..d1c379c 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java @@ -26,6 +26,7 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput; import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; import com.kingsrook.qqq.backend.core.model.data.QRecord; import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; @@ -153,4 +154,21 @@ class QuickSearchRuntimeTest extends BaseQuickSearchTest failFast.close(); } + + + @Test + void testStart_missingOptionalLibrary_degradedBoots_failFastThrows() + { + QCodeReference customizer = new QCodeReference(MissingLibraryTransportCustomizer.class); + + QuickSearchRuntime degraded = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withStartupMode(QuickSearchStartupMode.DEGRADED).withTransportCustomizer(customizer)); + assertThatCode(degraded::start).doesNotThrowAnyException(); + assertThat(degraded.isIndexReady()).isFalse(); + degraded.close(); + + QuickSearchRuntime failFast = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withTransportCustomizer(customizer)); + assertThatThrownBy(failFast::start).isInstanceOf(QException.class).hasMessageContaining("startupMode=DEGRADED").hasCauseInstanceOf(NoClassDefFoundError.class); + failFast.close(); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java index 713a289..ba28565 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java @@ -17,15 +17,19 @@ package com.kingsrook.qbits.quicksearch.listeners; +import java.util.ArrayList; import java.util.List; import com.kingsrook.qqq.backend.core.actions.QBackendTransaction; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; +import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeEvent; import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.logging.QCollectingLogger; +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.actions.tables.insert.InsertInput; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; @@ -35,6 +39,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QInstance; import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData; import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest; +import com.kingsrook.qbits.quicksearch.MissingLibraryTransportCustomizer; import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig; import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext; import com.kingsrook.qbits.quicksearch.QuickSearchRuntime; @@ -45,17 +50,23 @@ import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep; import com.kingsrook.qbits.quicksearch.publisher.IndexEvent; import com.kingsrook.qbits.quicksearch.publisher.IndexEventAction; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; +import org.apache.logging.log4j.Level; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.mockito.MockedConstruction; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /******************************************************************************* @@ -298,6 +309,147 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest + @Test + void testUpdate_reReadFails_recordsFailedEventsAfterCommit() throws QException + { + QBackendTransaction transaction = new QBackendTransaction(); + try(MockedConstruction ignored = failingQueries()) + { + assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7, 8))).doesNotThrowAnyException(); + } + + assertThat(queryFailedEvents()).isEmpty(); + transaction.commit(); + + List failed = queryFailedEvents(); + assertThat(failed).extracting(r -> r.getValueString("recordId")).containsExactlyInAnyOrder("7", "8"); + assertThat(failed).allSatisfy(r -> + { + assertThat(r.getValueString("tableName")).isEqualTo(TEST_ENTITY_TABLE); + assertThat(r.getValueString("action")).isEqualTo("INDEX"); + assertThat(r.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING); + assertThat(r.getValueString("errorMessage")).contains("possible-value backend unavailable"); + }); + verify(publisher, never()).publishIndexEvents(anyList()); + } + + + + @Test + void testUpdate_reReadFailsThenRollback_recordsNothing() throws QException + { + QBackendTransaction transaction = new QBackendTransaction(); + try(MockedConstruction ignored = failingQueries()) + { + new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7)); + } + transaction.rollback(); + + assertThat(queryFailedEvents()).isEmpty(); + } + + + + @Test + void testUpdate_reReadFailsWithoutTransaction_recordsImmediately() throws QException + { + try(MockedConstruction ignored = failingQueries()) + { + new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7)); + } + + assertThat(queryFailedEvents()).extracting(r -> r.getValueString("recordId")).containsExactly("7"); + } + + + + @Test + void testUpdate_reReadAndRecordingBothFail_logsAndDoesNotThrow() + { + QCollectingLogger collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRecordChangeListener.class); + try(MockedConstruction ignoredQueries = failingQueries(); + MockedConstruction ignoredInserts = mockConstruction(InsertAction.class, (mock, context) -> + when(mock.execute(any(InsertInput.class))).thenThrow(new QException("failed-event table unavailable")))) + { + assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7))).doesNotThrowAnyException(); + } + finally + { + QLogger.deactivateCollectingLoggerForClass(QuickSearchRecordChangeListener.class); + } + + assertThat(collectingLogger.getCollectedMessages()) + .anySatisfy(m -> + { + assertThat(m.getLevel()).isEqualTo(Level.WARN); + assertThat(m.getMessage()).contains("recording failed events for replay").contains(TEST_ENTITY_TABLE).doesNotContain("changed"); + }) + .anySatisfy(m -> + { + assertThat(m.getLevel()).isEqualTo(Level.ERROR); + assertThat(m.getMessage()).contains("Could not record failed index events"); + }); + } + + + + @Test + void testClientBuildFailsLinkage_recordsFailedEvents() throws QException + { + ////////////////////////////////////////////////////////////////// + // no host publisher: the first write builds the client lazily, // + // and a missing optional library fails linkage there // + ////////////////////////////////////////////////////////////////// + runtime.getConfig().setIndexEventPublisher(null); + runtime.getConfig().setTransportCustomizer(new QCodeReference(MissingLibraryTransportCustomizer.class)); + + Integer id = insert("widget", null); + + List failed = queryFailedEvents(); + assertThat(failed).hasSize(1); + assertThat(failed.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id)); + assertThat(failed.get(0).getValueString("action")).isEqualTo("INDEX"); + assertThat(failed.get(0).getValueString("errorMessage")).contains(MissingLibraryTransportCustomizer.MISSING_CLASS); + } + + + + @Test + void testPublishFailsLinkage_recordsFailedEvents() throws QException + { + doThrow(new NoClassDefFoundError("com/example/broker/Client")).when(publisher).publishDeleteEvents(anyList()); + Integer id = insert("widget", null); + delete(id); + + List failed = queryFailedEvents(); + assertThat(failed).hasSize(1); + assertThat(failed.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id)); + assertThat(failed.get(0).getValueString("action")).isEqualTo("DELETE"); + assertThat(failed.get(0).getValueString("errorMessage")).contains("com/example/broker/Client"); + } + + + + private MockedConstruction failingQueries() + { + return (mockConstruction(QueryAction.class, (mock, context) -> + when(mock.execute(any(QueryInput.class))).thenThrow(new QException("possible-value backend unavailable")))); + } + + + + private RecordChangeEvent updateEvent(QBackendTransaction transaction, Integer... ids) + { + List records = new ArrayList<>(); + for(Integer id : ids) + { + records.add(new QRecord().withValue("id", id).withValue("description", "changed")); + } + return (new RecordChangeEvent().withTableName(TEST_ENTITY_TABLE).withType(RecordChangeType.UPDATE).withRecords(records).withTransaction(transaction)); + } + + + private void delete(Integer id) throws QException { DeleteInput deleteInput = new DeleteInput(); -- 2.54.0 From dfac7c3c9ae90de57b6828bddbf221d07b5c76d2 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 6 Oct 2026 18:36:09 -0500 Subject: [PATCH 2/2] fix(#13): log a failed re-read at once and name a missing class at start The listener now also catches a LinkageError from the re-read and logs a warn with the table name and record count before deferring the failure rows to the commit, so a rollback still leaves a trace. A LinkageError at start names the missing class and the optional AWS SDK v2 dependencies instead of reporting OpenSearch as unreachable. Refs #13 --- .../qbits/quicksearch/QuickSearchRuntime.java | 17 ++++++- .../QuickSearchRecordChangeListener.java | 16 ++++-- .../quicksearch/QuickSearchRuntimeTest.java | 26 ++++++++-- .../QuickSearchRecordChangeListenerTest.java | 51 +++++++++++++++---- 4 files changed, 90 insertions(+), 20 deletions(-) diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java index dd5dd35..dd89bcf 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java @@ -340,7 +340,8 @@ public class QuickSearchRuntime ** Start the runtime per the configured startup mode: FAIL_FAST throws when ** the index cannot be prepared; DEGRADED logs and continues. A LinkageError ** (an optional library such as the AWS SDK missing while the client is - ** built) is handled the same way, so a DEGRADED host still boots. + ** built) follows the same modes, with a message naming the missing class, + ** so a DEGRADED host still boots. *******************************************************************************/ public void start() throws QException { @@ -348,7 +349,19 @@ public class QuickSearchRuntime { ensureIndexReady(); } - catch(Exception | LinkageError e) + catch(LinkageError e) + { + String message = "Quick Search could not load a class to build its OpenSearch client [" + e.getMessage() + + "]; authMode AWS_SIGV4 needs the optional AWS SDK v2 dependencies (apache-client, sts) on the classpath"; + if(config.getStartupMode() == QuickSearchStartupMode.DEGRADED) + { + LOG.warn(message + "; Quick Search is running degraded", e, logPair("indexName", config.getOpensearchIndexName())); + return; + } + + throw (new QException(message, e)); + } + catch(Exception e) { if(config.getStartupMode() == QuickSearchStartupMode.DEGRADED) { diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java index c35a57c..a357f09 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java @@ -122,11 +122,16 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, fetchCurrentRecords(event, primaryKeyField))); } - catch(Exception e) + catch(Exception | LinkageError e) { - //////////////////////////////////////////////////////////////////// - // record the updated keys, so basepull re-reads and indexes them // - //////////////////////////////////////////////////////////////////// + /////////////////////////////////////////////////////////////////////// + // log now: with a transaction, nothing else is logged until commit, // + // and a rollback (a transaction the failed re-read aborted, too) // + // would leave no trace. Then record the updated keys after commit, // + // so basepull re-reads and indexes them. // + /////////////////////////////////////////////////////////////////////// + LOG.warn("Could not re-read updated records for indexing; recording them for replay once the write commits", e, + logPair("tableName", tableName), logPair("recordCount", event.getRecords().size())); List failedEvents = toIndexEvents(tableName, primaryKeyField, event.getRecords()); runAfterCommit(event, () -> recordFailures(runtime, failedEvents, e)); return; @@ -441,7 +446,8 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* ** Updated records are "as the backend returned them", which for RDBMS is the ** sparse input. Re-read the full rows inside the same transaction, with - ** display values so possible-value labels are indexed. + ** display values so possible-value labels are indexed (which can run host + ** possible-value providers). *******************************************************************************/ private static List fetchCurrentRecords(RecordChangeEvent event, String primaryKeyField) throws QException { diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java index d1c379c..23687f8 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java @@ -22,6 +22,8 @@ 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.logging.QCollectingLogger; +import com.kingsrook.qqq.backend.core.logging.QLogger; 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; @@ -31,6 +33,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher; +import org.apache.logging.log4j.Level; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; @@ -161,13 +164,30 @@ class QuickSearchRuntimeTest extends BaseQuickSearchTest { QCodeReference customizer = new QCodeReference(MissingLibraryTransportCustomizer.class); - QuickSearchRuntime degraded = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withStartupMode(QuickSearchStartupMode.DEGRADED).withTransportCustomizer(customizer)); - assertThatCode(degraded::start).doesNotThrowAnyException(); + QuickSearchRuntime degraded = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withStartupMode(QuickSearchStartupMode.DEGRADED).withTransportCustomizer(customizer)); + QCollectingLogger collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRuntime.class); + try + { + assertThatCode(degraded::start).doesNotThrowAnyException(); + } + finally + { + QLogger.deactivateCollectingLoggerForClass(QuickSearchRuntime.class); + } assertThat(degraded.isIndexReady()).isFalse(); + assertThat(collectingLogger.getCollectedMessages()).anySatisfy(m -> + { + assertThat(m.getLevel()).isEqualTo(Level.WARN); + assertThat(m.getMessage()).contains(MissingLibraryTransportCustomizer.MISSING_CLASS).contains("AWS SDK v2").contains("apache-client, sts").doesNotContain("not reachable"); + }); degraded.close(); QuickSearchRuntime failFast = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withTransportCustomizer(customizer)); - assertThatThrownBy(failFast::start).isInstanceOf(QException.class).hasMessageContaining("startupMode=DEGRADED").hasCauseInstanceOf(NoClassDefFoundError.class); + assertThatThrownBy(failFast::start).isInstanceOf(QException.class) + .hasMessageContaining(MissingLibraryTransportCustomizer.MISSING_CLASS) + .hasMessageContaining("AWS SDK v2 dependencies (apache-client, sts)") + .hasMessageNotContaining("startupMode=DEGRADED") + .hasCauseInstanceOf(NoClassDefFoundError.class); failFast.close(); } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java index ba28565..76f6a28 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java @@ -313,7 +313,7 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest void testUpdate_reReadFails_recordsFailedEventsAfterCommit() throws QException { QBackendTransaction transaction = new QBackendTransaction(); - try(MockedConstruction ignored = failingQueries()) + try(MockedConstruction ignored = failingQueries(new QException("possible-value backend unavailable"))) { assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7, 8))).doesNotThrowAnyException(); } @@ -336,16 +336,42 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest @Test - void testUpdate_reReadFailsThenRollback_recordsNothing() throws QException + void testUpdate_reReadFailsThenRollback_logsCauseAndRecordsNothing() throws QException { - QBackendTransaction transaction = new QBackendTransaction(); - try(MockedConstruction ignored = failingQueries()) + QCollectingLogger collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRecordChangeListener.class); + QBackendTransaction transaction = new QBackendTransaction(); + try(MockedConstruction ignored = failingQueries(new QException("possible-value backend unavailable"))) { - new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7)); + new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7, 8)); + } + finally + { + QLogger.deactivateCollectingLoggerForClass(QuickSearchRecordChangeListener.class); } transaction.rollback(); assertThat(queryFailedEvents()).isEmpty(); + assertThat(collectingLogger.getCollectedMessages()).anySatisfy(m -> + { + assertThat(m.getLevel()).isEqualTo(Level.WARN); + assertThat(m.getMessage()).contains("Could not re-read updated records").contains(TEST_ENTITY_TABLE).contains("\"recordCount\":2").doesNotContain("changed"); + assertThat(m.getMessage()).contains("possible-value backend unavailable"); + }); + } + + + + @Test + void testUpdate_reReadFailsLinkage_recordsFailedEvents() throws QException + { + try(MockedConstruction ignored = failingQueries(new NoClassDefFoundError("com/example/pvs/Provider"))) + { + assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7))).doesNotThrowAnyException(); + } + + List failed = queryFailedEvents(); + assertThat(failed).extracting(r -> r.getValueString("recordId")).containsExactly("7"); + assertThat(failed.get(0).getValueString("errorMessage")).contains("com/example/pvs/Provider"); } @@ -353,7 +379,7 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest @Test void testUpdate_reReadFailsWithoutTransaction_recordsImmediately() throws QException { - try(MockedConstruction ignored = failingQueries()) + try(MockedConstruction ignored = failingQueries(new QException("possible-value backend unavailable"))) { new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7)); } @@ -367,7 +393,7 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest void testUpdate_reReadAndRecordingBothFail_logsAndDoesNotThrow() { QCollectingLogger collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRecordChangeListener.class); - try(MockedConstruction ignoredQueries = failingQueries(); + try(MockedConstruction ignoredQueries = failingQueries(new QException("possible-value backend unavailable")); MockedConstruction ignoredInserts = mockConstruction(InsertAction.class, (mock, context) -> when(mock.execute(any(InsertInput.class))).thenThrow(new QException("failed-event table unavailable")))) { @@ -382,7 +408,7 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest .anySatisfy(m -> { assertThat(m.getLevel()).isEqualTo(Level.WARN); - assertThat(m.getMessage()).contains("recording failed events for replay").contains(TEST_ENTITY_TABLE).doesNotContain("changed"); + assertThat(m.getMessage()).contains("Could not re-read updated records").contains(TEST_ENTITY_TABLE).doesNotContain("changed"); }) .anySatisfy(m -> { @@ -430,10 +456,15 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest - private MockedConstruction failingQueries() + private MockedConstruction failingQueries(Throwable failure) { + ///////////////////////////////////////////////////////// + // read the enabled flag first, so only the listener's // + // re-read meets the failing QueryAction // + ///////////////////////////////////////////////////////// + runtime.isTableEnabled(TEST_ENTITY_TABLE); return (mockConstruction(QueryAction.class, (mock, context) -> - when(mock.execute(any(QueryInput.class))).thenThrow(new QException("possible-value backend unavailable")))); + when(mock.execute(any(QueryInput.class))).thenThrow(failure))); } -- 2.54.0