standardize error logging for alerts

Ian-Nara · Ian-Nara · commit bf665a0f3a4e · 2025-12-08T16:44:43.000-07:00
diff --git a/src/main/java/com/uid2/optout/delta/DeltaProductionOrchestrator.java b/src/main/java/com/uid2/optout/delta/DeltaProductionOrchestrator.java
@@ -185,7 +185,7 @@ private boolean processWindow(SqsWindowReader.WindowReadResult windowResult,
         OptOutTrafficCalculator.TrafficStatus trafficStatus = this.trafficCalculator.calculateStatus(deltaMessages, queueAttributes, droppedMessages.size(), filteredAsTooRecentCount);
         
         if (trafficStatus == OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING) {
-            LOGGER.error("optout delta production has hit DELAYED_PROCESSING status, stopping production and setting manual override");
+            LOGGER.error("circuit_breaker_triggered: traffic spike detected, stopping production and setting manual override");
             manualOverrideService.setDelayedProcessing();
             return true;
         }
@@ -262,11 +262,11 @@ private boolean isJobTimedOut(long jobStartTime) {
         long elapsedTime = OptOutUtils.nowEpochSeconds() - jobStartTime;
         
         if (elapsedTime > 3600) { // 1 hour - log warning
-            LOGGER.error("delta production job has been running for {} seconds", elapsedTime);
+            LOGGER.error("delta_job_timeout: job has been running for {} seconds", elapsedTime);
         }
         
         if (elapsedTime > this.jobTimeoutSeconds) {
-            LOGGER.error("delta production job has been running for {} seconds (exceeds timeout of {}s)",
+            LOGGER.error("delta_job_timeout: job exceeded timeout, running for {} seconds (timeout: {}s)",
                     elapsedTime, this.jobTimeoutSeconds);
             return true;
         }
diff --git a/src/main/java/com/uid2/optout/delta/ManualOverrideService.java b/src/main/java/com/uid2/optout/delta/ManualOverrideService.java
@@ -62,7 +62,7 @@ public boolean setDelayedProcessing() {
             LOGGER.info("set manual override to DELAYED_PROCESSING: {}", overrideS3Path);
             return true;
         } catch (Exception e) {
-            LOGGER.error("error setting manual override: {}", overrideS3Path, e);
+            LOGGER.error("manual_override_error: failed to set override at {}", overrideS3Path, e);
             return false;
         }
     }
@@ -76,7 +76,7 @@ private String getOverrideValue() {
             JsonObject configJson = Utils.toJsonObject(inputStream);
             return configJson.getString(OVERRIDE_KEY, "");
         } catch (Exception e) {
-            LOGGER.error("no manual override found: {}", overrideS3Path);
+            LOGGER.error("manual_override_error: no manual override file found at {}", overrideS3Path);
             return "";
         }
     }
diff --git a/src/main/java/com/uid2/optout/delta/S3UploadService.java b/src/main/java/com/uid2/optout/delta/S3UploadService.java
@@ -72,7 +72,7 @@ public void uploadAndDeleteMessages(byte[] data, String s3Path, List<Message> me
                 onSuccess.onSuccess(messages.size());
             }
         } catch (Exception e) {
-            LOGGER.error("failed to upload to s3: path={}", s3Path, e);
+            LOGGER.error("s3_error: failed to upload delta or dropped requests to path={}", s3Path, e);
             throw new IOException("s3 upload failed: " + s3Path, e);
         }
         
diff --git a/src/main/java/com/uid2/optout/sqs/SqsBatchProcessor.java b/src/main/java/com/uid2/optout/sqs/SqsBatchProcessor.java
@@ -82,14 +82,14 @@ public BatchProcessingResult processBatch(List<Message> messageBatch, int batchN
         if (parsedBatch.size() < messageBatch.size()) {
             List<Message> invalidMessages = identifyInvalidMessages(messageBatch, parsedBatch);
             if (!invalidMessages.isEmpty()) {
-                LOGGER.error("found {} invalid messages in batch {}, deleting", invalidMessages.size(), batchNumber);
+                LOGGER.error("sqs_error: found {} invalid messages in batch {}, deleting", invalidMessages.size(), batchNumber);
                 SqsMessageOperations.deleteMessagesFromSqs(this.sqsClient, this.queueUrl, invalidMessages);
             }
         }
         
         // No valid messages after deleting corrupt ones, continue reading
         if (parsedBatch.isEmpty()) {
-            LOGGER.error("no valid messages in batch {}", batchNumber);
+            LOGGER.info("no valid messages in batch {} after removing invalid messages", batchNumber);
             return BatchProcessingResult.corruptMessagesDeleted();
         }
 
diff --git a/src/main/java/com/uid2/optout/sqs/SqsMessageOperations.java b/src/main/java/com/uid2/optout/sqs/SqsMessageOperations.java
@@ -136,7 +136,7 @@ public static List<Message> receiveMessagesFromSqs(
             return response.messages();
 
         } catch (Exception e) {
-            LOGGER.error("error receiving messages", e);
+            LOGGER.error("sqs_error: failed to receive messages", e);
             return new ArrayList<>();
         }
     }
@@ -174,7 +174,7 @@ public static void deleteMessagesFromSqs(SqsClient sqsClient, String queueUrl, L
                     DeleteMessageBatchResponse deleteResponse = sqsClient.deleteMessageBatch(deleteRequest);
 
                     if (!deleteResponse.failed().isEmpty()) {
-                        LOGGER.error("failed to delete {} messages", deleteResponse.failed().size());
+                        LOGGER.error("sqs_error: failed to delete {} messages", deleteResponse.failed().size());
                     } else {
                         totalDeleted += entries.size();
                     }
@@ -186,7 +186,7 @@ public static void deleteMessagesFromSqs(SqsClient sqsClient, String queueUrl, L
             LOGGER.info("deleted {} messages", totalDeleted);
 
         } catch (Exception e) {
-            LOGGER.error("error deleting messages", e);
+            LOGGER.error("sqs_error: exception during message deletion", e);
         }
     }
 }
diff --git a/src/main/java/com/uid2/optout/sqs/SqsMessageParser.java b/src/main/java/com/uid2/optout/sqs/SqsMessageParser.java
@@ -40,21 +40,21 @@ public static List<SqsParsedMessage> parseAndSortMessages(List<Message> messages
                 String phone = body.getString("phone");
 
                 if (identityHash == null || advertisingId == null) {
-                    LOGGER.error("invalid message format: {}", message.body());
+                    LOGGER.error("sqs_error: invalid message format: {}", message.body());
                     continue;
                 }
 
                 byte[] hashBytes = OptOutUtils.base64StringTobyteArray(identityHash);
                 byte[] idBytes = OptOutUtils.base64StringTobyteArray(advertisingId);
 
                 if (hashBytes == null || idBytes == null) {
-                    LOGGER.error("invalid base64 encoding");
+                    LOGGER.error("sqs_error: invalid base64 encoding");
                     continue;
                 }
 
                 parsedMessages.add(new SqsParsedMessage(message, hashBytes, idBytes, timestampSeconds, email, phone, clientIp, traceId));
             } catch (Exception e) {
-                LOGGER.error("error parsing message", e);
+                LOGGER.error("sqs_error: error parsing message", e);
             }
         }
 
@@ -73,7 +73,7 @@ public static List<SqsParsedMessage> parseAndSortMessages(List<Message> messages
     private static long extractTimestamp(Message message) {
         String sentTimestampStr = message.attributes().get(MessageSystemAttributeName.SENT_TIMESTAMP);
         if (sentTimestampStr == null) {
-            LOGGER.warn("message missing SentTimestamp, using current time");
+            LOGGER.info("message missing SentTimestamp, using current time");
             return OptOutUtils.nowEpochSeconds();
         }
         return Long.parseLong(sentTimestampStr) / 1000; // ms to seconds
diff --git a/src/main/java/com/uid2/optout/sqs/SqsWindowReader.java b/src/main/java/com/uid2/optout/sqs/SqsWindowReader.java
@@ -96,7 +96,7 @@ public WindowReadResult readWindow() {
         
         while (true) {
             if (windowMessages.size() >= maxMessagesPerWindow) {
-                LOGGER.warn("message limit exceeded: {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);
+                LOGGER.warn("high_message_volume: message limit exceeded while reading window, {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);
                 return WindowReadResult.messageLimitExceeded(windowMessages, currentWindowStart, rawMessagesRead);
             }
             
diff --git a/src/main/java/com/uid2/optout/traffic/OptOutTrafficCalculator.java b/src/main/java/com/uid2/optout/traffic/OptOutTrafficCalculator.java
@@ -139,10 +139,10 @@ public void reloadTrafficCalcConfig() throws MalformedTrafficCalcConfigException
                        this.evaluationWindowSeconds, this.baselineTraffic, this.thresholdMultiplier, ranges.size());
             
         } catch (MalformedTrafficCalcConfigException e) {
-            LOGGER.warn("failed to load traffic calc config, config is malformed: {}", trafficCalcConfigPath, e);
+            LOGGER.error("circuit_breaker_config_error: config is malformed, configPath={}", trafficCalcConfigPath, e);
             throw e;
         } catch (Exception e) {
-            LOGGER.warn("failed to load traffic calc config, config is malformed or missing: {}", trafficCalcConfigPath, e);
+            LOGGER.error("circuit_breaker_config_error: config is malformed or missing, configPath={}", trafficCalcConfigPath, e);
             throw new MalformedTrafficCalcConfigException("failed to load traffic calc config: " + e.getMessage());
         }
     }
@@ -163,12 +163,12 @@ List<List<Long>> parseAllowlistRanges(JsonObject config) throws MalformedTraffic
                         long end = rangeArray.getLong(1);
                         
                         if(start >= end) {
-                            LOGGER.error("invalid allowlist range: start must be less than end: [{}, {}]", start, end);
+                            LOGGER.error("circuit_breaker_config_error: allowlist range start must be less than end, range=[{}, {}]", start, end);
                             throw new MalformedTrafficCalcConfigException("invalid allowlist range at index " + i + ": start must be less than end");
                         }
 
                         if (end - start > 86400) {
-                            LOGGER.error("invalid allowlist range: range must be less than 24 hours: [{}, {}]", start, end);
+                            LOGGER.error("circuit_breaker_config_error: allowlist range must be less than 24 hours, range=[{}, {}]", start, end);
                             throw new MalformedTrafficCalcConfigException("invalid allowlist range at index " + i + ": range must be less than 24 hours");
                         }
                         
@@ -186,7 +186,7 @@ List<List<Long>> parseAllowlistRanges(JsonObject config) throws MalformedTraffic
                 long currentEnd = ranges.get(i).get(1);
                 long nextStart = ranges.get(i + 1).get(0);
                 if (currentEnd >= nextStart) {
-                    LOGGER.error("overlapping allowlist ranges detected: [{}, {}] overlaps with [{}, {}]",
+                    LOGGER.error("circuit_breaker_config_error: overlapping allowlist ranges, range=[{}, {}] overlaps with range=[{}, {}]",
                             ranges.get(i).get(0), currentEnd, nextStart, ranges.get(i + 1).get(1));
                     throw new MalformedTrafficCalcConfigException(
                             "overlapping allowlist ranges detected at indices " + i + " and " + (i + 1));
@@ -196,7 +196,7 @@ List<List<Long>> parseAllowlistRanges(JsonObject config) throws MalformedTraffic
         } catch (MalformedTrafficCalcConfigException e) {
             throw e;
         } catch (Exception e) {
-            LOGGER.error("failed to parse allowlist ranges", e);
+            LOGGER.error("circuit_breaker_config_error: failed to parse allowlist ranges", e);
             throw new MalformedTrafficCalcConfigException("failed to parse allowlist ranges: " + e.getMessage());
         }
         
@@ -227,8 +227,8 @@ public TrafficStatus calculateStatus(List<Message> sqsMessages, SqsMessageOperat
             List<String> deltaS3Paths = listDeltaFiles();
             
             if (deltaS3Paths.isEmpty()) {
-                LOGGER.warn("no delta files found in s3 with prefix: {}", s3DeltaPrefix);
-                return TrafficStatus.DEFAULT;
+                LOGGER.error("s3_error: no delta files found in s3 at prefix={}", s3DeltaPrefix);
+                throw new RuntimeException("no delta files found in s3 at prefix=" + s3DeltaPrefix);
             }
             
             // Find newest delta file timestamp for delta traffic window
@@ -326,8 +326,8 @@ public TrafficStatus calculateStatus(List<Message> sqsMessages, SqsMessageOperat
             return status;
             
         } catch (Exception e) {
-            LOGGER.error("error calculating traffic status", e);
-            return TrafficStatus.DEFAULT;
+            LOGGER.error("delta_job_failed: error calculating traffic status", e);
+            throw new RuntimeException("error calculating traffic status", e);
         }
     }
     
@@ -345,7 +345,7 @@ private long findNewestDeltaTimestamp(List<String> deltaS3Paths) throws IOExcept
         List<Long> timestamps = getTimestampsFromFile(newestDeltaPath);
         
         if (timestamps.isEmpty()) {
-            LOGGER.warn("newest delta file has no timestamps: {}", newestDeltaPath);
+            LOGGER.error("s3_error: newest delta file has no timestamps, path={}", newestDeltaPath);
             return System.currentTimeMillis() / 1000;
         }
         
@@ -372,7 +372,7 @@ private List<String> listDeltaFiles() {
             return deltaFiles;
                 
         } catch (Exception e) {
-            LOGGER.error("failed to list delta files from s3 with prefix: {}", s3DeltaPrefix, e);
+            LOGGER.error("s3_error: failed to list delta files at prefix={}", s3DeltaPrefix, e);
             return Collections.emptyList();
         }
     }
@@ -429,7 +429,7 @@ private List<Long> readTimestampsFromS3(String s3Path) throws IOException {
             
             return timestamps;
         } catch (Exception e) {
-            LOGGER.error("failed to read delta file from s3: {}", s3Path, e);
+            LOGGER.error("s3_error: failed to read delta file at path={}", s3Path, e);
             throw new IOException("failed to read delta file from s3: " + s3Path, e);
         }
     }
@@ -512,7 +512,7 @@ private Long extractTimestampFromMessage(Message msg) {
             try {
                 return Long.parseLong(sentTimestamp) / 1000;  // Convert ms to seconds
             } catch (NumberFormatException e) {
-                LOGGER.warn("invalid sentTimestamp: {}", sentTimestamp);
+                LOGGER.error("sqs_error: invalid sentTimestamp, messageId={}, sentTimestamp={}", msg.messageId(), sentTimestamp);
             }
         }
         
@@ -590,23 +590,38 @@ private void evictOldCacheEntries(long cutoffTimestamp) {
     }
     
     /**
-     * Determine traffic status based on current vs past counts
+     * Determine traffic status based on current vs baseline traffic.
+     * Logs warnings at 50%, 75%, and 90% of the circuit breaker threshold.
      */
     TrafficStatus determineStatus(int sumCurrent, int baselineTraffic) {
         if (baselineTraffic == 0 || thresholdMultiplier == 0) {
-            // Avoid division by zero - if no baseline traffic, return DEFAULT status
-            LOGGER.warn("baselineTraffic is 0 or thresholdMultiplier is 0, returning default status");
-            return TrafficStatus.DEFAULT;
+            LOGGER.error("circuit_breaker_config_error: baselineTraffic is 0 or thresholdMultiplier is 0");
+            throw new RuntimeException("invalid circuit breaker config: baselineTraffic=" + baselineTraffic + ", thresholdMultiplier=" + thresholdMultiplier);
         }
         
-        if (sumCurrent >= thresholdMultiplier * baselineTraffic) {
-            LOGGER.error("delayed_processing threshold breached: sumCurrent={}, thresholdMultiplier={}, baselineTraffic={}", 
-                       sumCurrent, thresholdMultiplier, baselineTraffic);
+        int threshold = thresholdMultiplier * baselineTraffic;
+        double thresholdPercent = (double) sumCurrent / threshold * 100;
+        
+        // Log warnings at increasing thresholds before circuit breaker triggers
+        if (thresholdPercent >= 90.0) {
+            LOGGER.warn("high_message_volume: 90% of threshold reached, sumCurrent={}, threshold={} ({}x{}), thresholdPercent={}%", 
+                       sumCurrent, threshold, thresholdMultiplier, baselineTraffic, String.format("%.1f", thresholdPercent));
+        } else if (thresholdPercent >= 75.0) {
+            LOGGER.warn("high_message_volume: 75% of threshold reached, sumCurrent={}, threshold={} ({}x{}), thresholdPercent={}%", 
+                       sumCurrent, threshold, thresholdMultiplier, baselineTraffic, String.format("%.1f", thresholdPercent));
+        } else if (thresholdPercent >= 50.0) {
+            LOGGER.warn("high_message_volume: 50% of threshold reached, sumCurrent={}, threshold={} ({}x{}), thresholdPercent={}%", 
+                       sumCurrent, threshold, thresholdMultiplier, baselineTraffic, String.format("%.1f", thresholdPercent));
+        }
+        
+        if (sumCurrent >= threshold) {
+            LOGGER.error("circuit_breaker_triggered: traffic threshold breached, sumCurrent={}, threshold={} ({}x{})", 
+                       sumCurrent, threshold, thresholdMultiplier, baselineTraffic);
             return TrafficStatus.DELAYED_PROCESSING;
         }
         
-        LOGGER.info("traffic within normal range: sumCurrent={}, thresholdMultiplier={}, baselineTraffic={}", 
-                   sumCurrent, thresholdMultiplier, baselineTraffic);
+        LOGGER.info("traffic within normal range: sumCurrent={}, threshold={} ({}x{}), thresholdPercent={}%", 
+                   sumCurrent, threshold, thresholdMultiplier, baselineTraffic, String.format("%.1f", thresholdPercent));
         return TrafficStatus.DEFAULT;
     }
     
diff --git a/src/main/java/com/uid2/optout/traffic/OptOutTrafficFilter.java b/src/main/java/com/uid2/optout/traffic/OptOutTrafficFilter.java
diff --git a/src/main/java/com/uid2/optout/vertx/OptOutSqsLogProducer.java b/src/main/java/com/uid2/optout/vertx/OptOutSqsLogProducer.java
diff --git a/src/test/java/com/uid2/optout/traffic/OptOutTrafficCalculatorTest.java b/src/test/java/com/uid2/optout/traffic/OptOutTrafficCalculatorTest.java
diff --git a/src/test/java/com/uid2/optout/vertx/OptOutSqsLogProducerTest.java b/src/test/java/com/uid2/optout/vertx/OptOutSqsLogProducerTest.java

Original file line number	Diff line number	Diff line change
`@@ -185,7 +185,7 @@ private boolean processWindow(SqsWindowReader.WindowReadResult windowResult,`
`185`	`185`	`OptOutTrafficCalculator.TrafficStatus trafficStatus = this.trafficCalculator.calculateStatus(deltaMessages, queueAttributes, droppedMessages.size(), filteredAsTooRecentCount);`
`186`	`186`
`187`	`187`	`if (trafficStatus == OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING) {`
`188`		`- LOGGER.error("optout delta production has hit DELAYED_PROCESSING status, stopping production and setting manual override");`
	`188`	`+ LOGGER.error("circuit_breaker_triggered: traffic spike detected, stopping production and setting manual override");`
`189`	`189`	`manualOverrideService.setDelayedProcessing();`
`190`	`190`	`return true;`
`191`	`191`	`}`
`@@ -262,11 +262,11 @@ private boolean isJobTimedOut(long jobStartTime) {`
`262`	`262`	`long elapsedTime = OptOutUtils.nowEpochSeconds() - jobStartTime;`
`263`	`263`
`264`	`264`	`if (elapsedTime > 3600) { // 1 hour - log warning`
`265`		`- LOGGER.error("delta production job has been running for {} seconds", elapsedTime);`
	`265`	`+ LOGGER.error("delta_job_timeout: job has been running for {} seconds", elapsedTime);`
`266`	`266`	`}`
`267`	`267`
`268`	`268`	`if (elapsedTime > this.jobTimeoutSeconds) {`
`269`		`- LOGGER.error("delta production job has been running for {} seconds (exceeds timeout of {}s)",`
	`269`	`+ LOGGER.error("delta_job_timeout: job exceeded timeout, running for {} seconds (timeout: {}s)",`
`270`	`270`	`elapsedTime, this.jobTimeoutSeconds);`
`271`	`271`	`return true;`
`272`	`272`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ public boolean setDelayedProcessing() {`
`62`	`62`	`LOGGER.info("set manual override to DELAYED_PROCESSING: {}", overrideS3Path);`
`63`	`63`	`return true;`
`64`	`64`	`} catch (Exception e) {`
`65`		`- LOGGER.error("error setting manual override: {}", overrideS3Path, e);`
	`65`	`+ LOGGER.error("manual_override_error: failed to set override at {}", overrideS3Path, e);`
`66`	`66`	`return false;`
`67`	`67`	`}`
`68`	`68`	`}`
`@@ -76,7 +76,7 @@ private String getOverrideValue() {`
`76`	`76`	`JsonObject configJson = Utils.toJsonObject(inputStream);`
`77`	`77`	`return configJson.getString(OVERRIDE_KEY, "");`
`78`	`78`	`} catch (Exception e) {`
`79`		`- LOGGER.error("no manual override found: {}", overrideS3Path);`
	`79`	`+ LOGGER.error("manual_override_error: no manual override file found at {}", overrideS3Path);`
`80`	`80`	`return "";`
`81`	`81`	`}`
`82`	`82`	`}`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ public void uploadAndDeleteMessages(byte[] data, String s3Path, List<Message> me`
`72`	`72`	`onSuccess.onSuccess(messages.size());`
`73`	`73`	`}`
`74`	`74`	`} catch (Exception e) {`
`75`		`- LOGGER.error("failed to upload to s3: path={}", s3Path, e);`
	`75`	`+ LOGGER.error("s3_error: failed to upload delta or dropped requests to path={}", s3Path, e);`
`76`	`76`	`throw new IOException("s3 upload failed: " + s3Path, e);`
`77`	`77`	`}`
`78`	`78`
Original file line number	Diff line number	Diff line change
`@@ -82,14 +82,14 @@ public BatchProcessingResult processBatch(List<Message> messageBatch, int batchN`
`82`	`82`	`if (parsedBatch.size() < messageBatch.size()) {`
`83`	`83`	`List<Message> invalidMessages = identifyInvalidMessages(messageBatch, parsedBatch);`
`84`	`84`	`if (!invalidMessages.isEmpty()) {`
`85`		`- LOGGER.error("found {} invalid messages in batch {}, deleting", invalidMessages.size(), batchNumber);`
	`85`	`+ LOGGER.error("sqs_error: found {} invalid messages in batch {}, deleting", invalidMessages.size(), batchNumber);`
`86`	`86`	`SqsMessageOperations.deleteMessagesFromSqs(this.sqsClient, this.queueUrl, invalidMessages);`
`87`	`87`	`}`
`88`	`88`	`}`
`89`	`89`
`90`	`90`	`// No valid messages after deleting corrupt ones, continue reading`
`91`	`91`	`if (parsedBatch.isEmpty()) {`
`92`		`- LOGGER.error("no valid messages in batch {}", batchNumber);`
	`92`	`+ LOGGER.info("no valid messages in batch {} after removing invalid messages", batchNumber);`
`93`	`93`	`return BatchProcessingResult.corruptMessagesDeleted();`
`94`	`94`	`}`
`95`	`95`
Original file line number	Diff line number	Diff line change
`@@ -136,7 +136,7 @@ public static List<Message> receiveMessagesFromSqs(`
`136`	`136`	`return response.messages();`
`137`	`137`
`138`	`138`	`} catch (Exception e) {`
`139`		`- LOGGER.error("error receiving messages", e);`
	`139`	`+ LOGGER.error("sqs_error: failed to receive messages", e);`
`140`	`140`	`return new ArrayList<>();`
`141`	`141`	`}`
`142`	`142`	`}`
`@@ -174,7 +174,7 @@ public static void deleteMessagesFromSqs(SqsClient sqsClient, String queueUrl, L`
`174`	`174`	`DeleteMessageBatchResponse deleteResponse = sqsClient.deleteMessageBatch(deleteRequest);`
`175`	`175`
`176`	`176`	`if (!deleteResponse.failed().isEmpty()) {`
`177`		`- LOGGER.error("failed to delete {} messages", deleteResponse.failed().size());`
	`177`	`+ LOGGER.error("sqs_error: failed to delete {} messages", deleteResponse.failed().size());`
`178`	`178`	`} else {`
`179`	`179`	`totalDeleted += entries.size();`
`180`	`180`	`}`
`@@ -186,7 +186,7 @@ public static void deleteMessagesFromSqs(SqsClient sqsClient, String queueUrl, L`
`186`	`186`	`LOGGER.info("deleted {} messages", totalDeleted);`
`187`	`187`
`188`	`188`	`} catch (Exception e) {`
`189`		`- LOGGER.error("error deleting messages", e);`
	`189`	`+ LOGGER.error("sqs_error: exception during message deletion", e);`
`190`	`190`	`}`
`191`	`191`	`}`
`192`	`192`	`}`
Original file line number	Diff line number	Diff line change
`@@ -40,21 +40,21 @@ public static List<SqsParsedMessage> parseAndSortMessages(List<Message> messages`
`40`	`40`	`String phone = body.getString("phone");`
`41`	`41`
`42`	`42`	`if (identityHash == null \|\| advertisingId == null) {`
`43`		`- LOGGER.error("invalid message format: {}", message.body());`
	`43`	`+ LOGGER.error("sqs_error: invalid message format: {}", message.body());`
`44`	`44`	`continue;`
`45`	`45`	`}`
`46`	`46`
`47`	`47`	`byte[] hashBytes = OptOutUtils.base64StringTobyteArray(identityHash);`
`48`	`48`	`byte[] idBytes = OptOutUtils.base64StringTobyteArray(advertisingId);`
`49`	`49`
`50`	`50`	`if (hashBytes == null \|\| idBytes == null) {`
`51`		`- LOGGER.error("invalid base64 encoding");`
	`51`	`+ LOGGER.error("sqs_error: invalid base64 encoding");`
`52`	`52`	`continue;`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`parsedMessages.add(new SqsParsedMessage(message, hashBytes, idBytes, timestampSeconds, email, phone, clientIp, traceId));`
`56`	`56`	`} catch (Exception e) {`
`57`		`- LOGGER.error("error parsing message", e);`
	`57`	`+ LOGGER.error("sqs_error: error parsing message", e);`
`58`	`58`	`}`
`59`	`59`	`}`
`60`	`60`
`@@ -73,7 +73,7 @@ public static List<SqsParsedMessage> parseAndSortMessages(List<Message> messages`
`73`	`73`	`private static long extractTimestamp(Message message) {`
`74`	`74`	`String sentTimestampStr = message.attributes().get(MessageSystemAttributeName.SENT_TIMESTAMP);`
`75`	`75`	`if (sentTimestampStr == null) {`
`76`		`- LOGGER.warn("message missing SentTimestamp, using current time");`
	`76`	`+ LOGGER.info("message missing SentTimestamp, using current time");`
`77`	`77`	`return OptOutUtils.nowEpochSeconds();`
`78`	`78`	`}`
`79`	`79`	`return Long.parseLong(sentTimestampStr) / 1000; // ms to seconds`
Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,7 @@ public WindowReadResult readWindow() {`
`96`	`96`
`97`	`97`	`while (true) {`
`98`	`98`	`if (windowMessages.size() >= maxMessagesPerWindow) {`
`99`		`- LOGGER.warn("message limit exceeded: {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);`
	`99`	`+ LOGGER.warn("high_message_volume: message limit exceeded while reading window, {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);`
`100`	`100`	`return WindowReadResult.messageLimitExceeded(windowMessages, currentWindowStart, rawMessagesRead);`
`101`	`101`	`}`
`102`	`102`