failuresByMessage) {
+ return failuresByMessage.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ }
+
static String getMatchingReservedIdIfAny(final String id) {
for (final String reservedId : RESERVED_IDS) {
if (id.contains(reservedId)) {
diff --git a/dotCMS/src/main/java/com/dotmarketing/util/Logger.java b/dotCMS/src/main/java/com/dotmarketing/util/Logger.java
index d2708664c261..80b8e9e53e98 100644
--- a/dotCMS/src/main/java/com/dotmarketing/util/Logger.java
+++ b/dotCMS/src/main/java/com/dotmarketing/util/Logger.java
@@ -244,10 +244,38 @@ public static void warnEvery(final Class cl, final String messageKey, final Stri
}
+ /**
+ * this method will print the message at ERROR level at most once per millis set, keyed by
+ * {@code messageKey}: a repeating condition (e.g. every bulk batch being rejected for the same
+ * reason) is reported without flooding the log, while a different key is reported
+ * immediately. Mirrors {@link #warnEvery(Class, String, String, int)} at ERROR level.
+ *
+ * @param cl the class to attribute the log entry to
+ * @param messageKey identity of the condition; the same key is throttled together
+ * @param message the message to print
+ * @param errorEveryMillis minimum millis between two entries for this key
+ */
+ public static void errorEvery(final Class cl, final String messageKey, final String message,
+ final int errorEveryMillis) {
+
+ if (UtilMethods.isEmpty(messageKey)) {
+ return;
+ }
+ final org.apache.logging.log4j.Logger logger = loadLogger(cl);
+
+ final Long hash = Long.valueOf(Objects.hashCode(messageKey.intern()));
+ final Long expireWhen = logMap.get().get(hash);
+
+ if (expireWhen == null || expireWhen < System.currentTimeMillis()) {
+ logMap.get().put(hash, System.currentTimeMillis() + errorEveryMillis, true);
+ logger.error(message + " (log every " + errorEveryMillis + "ms)");
+ }
+ }
+
/**
* this method will print the message at WARN level every millis set and print the message plus
* whole stack trace if at DEGUG level
- *
+ *
* @param cl
* @param message
* @param ex
diff --git a/dotCMS/src/test/java/com/dotcms/content/index/opensearch/OSIndexAPIImplConnectionClassifyTest.java b/dotCMS/src/test/java/com/dotcms/content/index/opensearch/OSIndexAPIImplConnectionClassifyTest.java
index da4386db2c1d..f1bb7689e48f 100644
--- a/dotCMS/src/test/java/com/dotcms/content/index/opensearch/OSIndexAPIImplConnectionClassifyTest.java
+++ b/dotCMS/src/test/java/com/dotcms/content/index/opensearch/OSIndexAPIImplConnectionClassifyTest.java
@@ -83,6 +83,35 @@ public void securityExceptionStatus403_isAuthForbidden() {
+ " reason=no permissions for [indices:admin/create]] status: 403"))));
}
+ /**
+ * A rejected bulk item never carries a status code: the client reports the cluster's
+ * refusal as plain text on the item. Without matching the security plugin's own wording, a
+ * permission problem that rejects every shadow write would be classified as
+ * {@link ConnectionFailureKind#UNKNOWN} and never escalated (issue #36222 follow-up).
+ */
+ @Test
+ public void bulkItemSecurityException_withoutStatusCode_isAuthForbidden() {
+ assertEquals(ConnectionFailureKind.AUTH_FORBIDDEN,
+ OSIndexAPIImpl.classifyFailureMessage(
+ "security_exception: no permissions for [indices:data/write/bulk[s],"
+ + " indices:data/write/index] and User [name=non-admin,"
+ + " backend_roles=[], requestedTenant=null]"));
+ }
+
+ @Test
+ public void bulkItemMappingFailure_isNotClassified() {
+ // A per-document problem must stay UNKNOWN: it is not a migration blocker, and reporting it
+ // as one would escalate every malformed contentlet.
+ assertEquals(ConnectionFailureKind.UNKNOWN,
+ OSIndexAPIImpl.classifyFailureMessage(
+ "mapper_parsing_exception: failed to parse field [myNumber] of type [long]"));
+ }
+
+ @Test
+ public void nullFailureMessage_isNotClassified() {
+ assertEquals(ConnectionFailureKind.UNKNOWN, OSIndexAPIImpl.classifyFailureMessage(null));
+ }
+
/**
* A dotCMS wrapper message embeds the physical index name, and a {@code _yyyyMMddHHmmss}
* timestamp regularly contains the digits 403 or 401 (here 12:04:03). Matching the status code
diff --git a/dotCMS/src/test/java/com/dotmarketing/common/reindex/BulkProcessorListenerShadowFailureTest.java b/dotCMS/src/test/java/com/dotmarketing/common/reindex/BulkProcessorListenerShadowFailureTest.java
new file mode 100644
index 000000000000..9d7165a3d3c9
--- /dev/null
+++ b/dotCMS/src/test/java/com/dotmarketing/common/reindex/BulkProcessorListenerShadowFailureTest.java
@@ -0,0 +1,219 @@
+package com.dotmarketing.common.reindex;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.dotcms.content.index.IndexTag;
+import com.dotcms.content.index.domain.IndexBulkItemResult;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.Test;
+
+/**
+ * Unit tests for how {@link BulkProcessorListener} reports failures of the OpenSearch shadow
+ * bulk leg in dual-write phases (issue #36222 follow-up).
+ *
+ * What this covers
+ * The listener used to log one WARN per failed item and nothing else. When the OS user loses a
+ * permission, every document in every batch is rejected with the same message, which produced
+ * hundreds of thousands of identical lines and — worse — no signal at all that the shadow store had
+ * stopped receiving writes altogether. QA hit exactly that on TC-056: ~900 identical warnings in one
+ * minute, and a reindex that looked successful while OpenSearch received nothing.
+ *
+ * Both helpers under test are pure functions, so no container, cluster or config is needed. The
+ * classification of the rejection text itself lives in the OpenSearch adapter and is exercised by
+ * {@code OSIndexAPIImplConnectionClassifyTest}; here we only assert the escalation policy built on
+ * top of it.
+ *
+ *
+ * ./mvnw test -pl :dotcms-core -Dmaven.build.cache.enabled=false \
+ * -Dtest=BulkProcessorListenerShadowFailureTest
+ *
+ *
+ * @author Fabrizzio Araya
+ */
+public class BulkProcessorListenerShadowFailureTest {
+
+ /** Verbatim rejection emitted by the OpenSearch security plugin for a denied bulk write. */
+ private static final String SECURITY_EXCEPTION =
+ "security_exception: no permissions for [indices:data/write/bulk[s],"
+ + " indices:data/write/index] and User [name=non-admin, backend_roles=[],"
+ + " requestedTenant=null]";
+
+ /** A per-document problem: the batch is healthy, this one contentlet is not. */
+ private static final String MAPPER_PARSING_EXCEPTION =
+ "mapper_parsing_exception: failed to parse field [myNumber] of type [long]";
+
+ private static IndexBulkItemResult failed(final String id, final String message) {
+ return IndexBulkItemResult.builder().id(id).failed(true).failureMessage(message).build();
+ }
+
+ private static IndexBulkItemResult succeeded(final String id) {
+ return IndexBulkItemResult.builder().id(id).failed(false).build();
+ }
+
+ private static List allFailedWith(final int count, final String message) {
+ return IntStream.range(0, count)
+ .mapToObj(i -> failed("id-" + i, message))
+ .collect(Collectors.toList());
+ }
+
+ // ---- summarizeFailures ----------------------------------------------------------------------
+
+ @Test
+ public void identicalFailures_areCollapsedIntoOneEntryWithItsCount() {
+ // The whole point: one log line for 185 rejections, not 185 lines.
+ final Map summary =
+ BulkProcessorListener.summarizeFailures(allFailedWith(185, SECURITY_EXCEPTION));
+
+ assertEquals("Identical rejections must collapse to a single entry", 1, summary.size());
+ assertEquals(Long.valueOf(185L), summary.get(SECURITY_EXCEPTION));
+ }
+
+ @Test
+ public void successfulItems_areNotCounted() {
+ final List results = List.of(
+ succeeded("ok-1"),
+ failed("bad-1", MAPPER_PARSING_EXCEPTION),
+ succeeded("ok-2"));
+
+ final Map summary = BulkProcessorListener.summarizeFailures(results);
+
+ assertEquals(1, summary.size());
+ assertEquals(Long.valueOf(1L), summary.get(MAPPER_PARSING_EXCEPTION));
+ }
+
+ @Test
+ public void distinctFailures_areKeptApart() {
+ final List results = List.of(
+ failed("bad-1", SECURITY_EXCEPTION),
+ failed("bad-2", MAPPER_PARSING_EXCEPTION),
+ failed("bad-3", SECURITY_EXCEPTION));
+
+ final Map summary = BulkProcessorListener.summarizeFailures(results);
+
+ assertEquals(2, summary.size());
+ assertEquals(Long.valueOf(2L), summary.get(SECURITY_EXCEPTION));
+ assertEquals(Long.valueOf(1L), summary.get(MAPPER_PARSING_EXCEPTION));
+ }
+
+ @Test
+ public void missingFailureMessage_doesNotProduceANullKey() {
+ final Map summary =
+ BulkProcessorListener.summarizeFailures(List.of(failed("bad-1", null)));
+
+ assertEquals(Long.valueOf(1L),
+ summary.get(BulkProcessorListener.NO_FAILURE_MESSAGE));
+ }
+
+ @Test
+ public void healthyBatch_producesNothing() {
+ assertTrue(BulkProcessorListener.summarizeFailures(
+ List.of(succeeded("ok-1"), succeeded("ok-2"))).isEmpty());
+ }
+
+ // ---- systemicFailureEscalation -------------------------------------------------------------
+
+ @Test
+ public void wholeBatchRejectedByPermissions_escalates() {
+ final List results = allFailedWith(185, SECURITY_EXCEPTION);
+
+ final Optional escalation = BulkProcessorListener.systemicFailureEscalation(
+ IndexTag.OS, results.size(), BulkProcessorListener.summarizeFailures(results));
+
+ assertTrue("A batch rejected in full for a permission problem must escalate",
+ escalation.isPresent());
+ final String message = escalation.get();
+ assertTrue("The cause must be named so the operator knows what to fix: " + message,
+ message.contains("AUTH_FORBIDDEN"));
+ assertTrue("The batch size must be reported: " + message, message.contains("185"));
+ assertTrue("The consequence — do not promote a diverging store — must be stated: " + message,
+ message.contains("must not be promoted"));
+ assertTrue("The verbatim rejection must be kept for support: " + message,
+ message.contains(SECURITY_EXCEPTION));
+ }
+
+ @Test
+ public void partiallyRejectedBatch_doesNotEscalate() {
+ // One rejected document out of many is a content problem, not a migration blocker — even
+ // when its message would classify as systemic on its own.
+ final List results = List.of(
+ succeeded("ok-1"),
+ succeeded("ok-2"),
+ failed("bad-1", SECURITY_EXCEPTION));
+
+ assertFalse(BulkProcessorListener.systemicFailureEscalation(IndexTag.OS, results.size(),
+ BulkProcessorListener.summarizeFailures(results)).isPresent());
+ }
+
+ @Test
+ public void wholeBatchRejectedByAMappingProblem_doesNotEscalate() {
+ // Every item failing is not enough: a batch of documents that share a broken field is a
+ // content problem, and halting the operator on it would be noise.
+ final List results = allFailedWith(10, MAPPER_PARSING_EXCEPTION);
+
+ assertFalse("An unclassifiable cause must not be reported as a systemic failure",
+ BulkProcessorListener.systemicFailureEscalation(IndexTag.OS, results.size(),
+ BulkProcessorListener.summarizeFailures(results)).isPresent());
+ }
+
+ @Test
+ public void mixedCauses_escalateOnTheDominantOne() {
+ // Realistic shape of a denied batch: the permission rejection dominates, one document also
+ // happens to be malformed. The escalation must name the permission problem.
+ final List results = List.of(
+ failed("bad-1", SECURITY_EXCEPTION),
+ failed("bad-2", SECURITY_EXCEPTION),
+ failed("bad-3", MAPPER_PARSING_EXCEPTION));
+
+ final Optional escalation = BulkProcessorListener.systemicFailureEscalation(
+ IndexTag.OS, results.size(), BulkProcessorListener.summarizeFailures(results));
+
+ assertTrue(escalation.isPresent());
+ assertTrue(escalation.get().contains("AUTH_FORBIDDEN"));
+ }
+
+ @Test
+ public void emptyBatch_doesNotEscalate() {
+ assertFalse(BulkProcessorListener.systemicFailureEscalation(
+ IndexTag.OS, 0, Map.of()).isPresent());
+ }
+
+ // ---- containment of the reporting itself -----------------------------------------------------
+
+ @Test
+ public void aFailureWhileReportingIsContained_andNeverReachesTheCaller() {
+ // Reporting a shadow failure must not become a failure. The OpenSearch adapter calls this
+ // callback inside its own try, so a throw here comes back as afterBulk(Throwable) and is
+ // logged as "Bulk process failed entirely" — a reporting defect would then look exactly like
+ // the shadow store having stopped accepting writes, the signal used to decide whether the
+ // store may be promoted.
+ //
+ // Error, not Exception, on purpose: the escalation path initialises OSIndexAPIImpl, so the
+ // realistic failure is a LinkageError, which neither the adapter's catch(Exception) nor
+ // CompositeBulkProcessor.close()'s shadow-isolation branch would contain.
+ final IndexBulkItemResult poisoned = new IndexBulkItemResult() {
+ @Override
+ public String id() {
+ return "boom";
+ }
+
+ @Override
+ public boolean failed() {
+ throw new NoClassDefFoundError("simulated broken static initialiser");
+ }
+
+ @Override
+ public String failureMessage() {
+ return null;
+ }
+ };
+
+ BulkProcessorListener.forShadowProvider(IndexTag.OS).afterBulk(1L, List.of(poisoned));
+ // Reaching this line is the assertion: the callback returned instead of propagating.
+ }
+}