Add NOISY marker to event processing error messages

Signed-off-by: Hai Yan <[email protected]>

Author: oeyh

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java

@@ -53,6 +53,8 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
+
 public class BinlogEventListener implements BinaryLogClient.EventListener {
 
     private static final Logger LOG = LoggerFactory.getLogger(BinlogEventListener.class);
@@ -464,7 +466,7 @@ private void handleEventAndErrors(com.github.shyiko.mysql.binlog.event.Event eve
         try {
             eventProcessingTimer.record(() -> function.accept(event));
         } catch (Exception e) {
-            LOG.error("Failed to process change event of type {}", event.getHeader().getEventType(), e);
+            LOG.error(NOISY, "Failed to process change event of type {}", event.getHeader().getEventType(), e);
             eventProcessingErrorCounter.increment();
         }
     }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationEventProcessor.java

@@ -45,6 +45,8 @@
 import java.util.Set;
 import java.util.function.Consumer;
 
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
+
 public class LogicalReplicationEventProcessor {
 
     enum TupleDataType {
@@ -493,7 +495,7 @@ private void handleMessageWithRetries(ByteBuffer message, Consumer<ByteBuffer> f
                 eventProcessingTimer.record(() -> function.accept(message));
                 return;
             } catch (Exception e) {
-                LOG.warn("Error when processing change event of type {}, will retry", messageType, e);
+                LOG.warn(NOISY, "Error when processing change event of type {}, will retry", messageType, e);
                 applyBackoff();
             }
             retry++;