diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java index 94c22da..dd89bcf 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java @@ -338,7 +338,10 @@ 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) follows the same modes, with a message naming the missing class, + ** so a DEGRADED host still boots. *******************************************************************************/ public void start() throws QException { @@ -346,6 +349,18 @@ public class QuickSearchRuntime { ensureIndexReady(); } + 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 990f884..a357f09 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,27 @@ 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 | LinkageError e) + { + /////////////////////////////////////////////////////////////////////// + // 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; + } + } case DELETE -> deleteEvents.addAll(toDeleteEvents(tableName, primaryKeyField, event.getRecords())); } @@ -122,15 +145,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 +193,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 +205,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 +218,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { publisher.publishIndexEvents(indexEvents); } - catch(Exception e) + catch(Exception | LinkageError e) { recordFailures(runtime, indexEvents, e); } @@ -195,7 +230,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte { publisher.publishDeleteEvents(deleteEvents); } - catch(Exception e) + catch(Exception | LinkageError e) { recordFailures(runtime, deleteEvents, e); } @@ -205,9 +240,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 +250,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 @@ -411,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/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..23687f8 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntimeTest.java @@ -22,14 +22,18 @@ 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; 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; 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; @@ -153,4 +157,38 @@ 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)); + 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(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 713a289..76f6a28 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,178 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest + @Test + void testUpdate_reReadFails_recordsFailedEventsAfterCommit() throws QException + { + QBackendTransaction transaction = new QBackendTransaction(); + try(MockedConstruction ignored = failingQueries(new QException("possible-value backend unavailable"))) + { + 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_logsCauseAndRecordsNothing() throws QException + { + 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, 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"); + } + + + + @Test + void testUpdate_reReadFailsWithoutTransaction_recordsImmediately() throws QException + { + try(MockedConstruction ignored = failingQueries(new QException("possible-value backend unavailable"))) + { + 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(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")))) + { + 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("Could not re-read updated records").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(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(failure))); + } + + + + 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();