Updated estimate memory of a batch to log an error if one of any event fields is of unrecognized type

Author: andsel

logstash-core/src/main/java/org/logstash/ConvertedMap.java

@@ -271,6 +271,10 @@ private long estimateMemory(Object o) {
             return Double.BYTES;
         }
 
-        throw new RuntimeException("Unsupported type in estimating memory size of event: " + o.getClass());
+        throw new IllegalArgumentException(
+                "Unsupported type encountered in estimateMemory: " + o.getClass().getName() +
+                        ". Please ensure all objects passed to estimateMemory are of supported types. " +
+                        "Refer to the ConvertedMap.estimateMemory method for the list of supported types."
+        );
     }
 }

logstash-core/src/main/java/org/logstash/execution/QueueReadClientBatchMetrics.java

@@ -1,5 +1,7 @@
 package org.logstash.execution;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.jruby.runtime.ThreadContext;
 import org.logstash.ackedqueue.QueueFactoryExt;
 import org.logstash.ext.JrubyEventExtLibrary;
@@ -11,6 +13,9 @@
 import static org.logstash.instrument.metrics.MetricKeys.*;
 
 class QueueReadClientBatchMetrics {
+
+    private static final Logger LOG = LogManager.getLogger(QueueReadClientBatchMetrics.class);
+
     private final QueueFactoryExt.BatchMetricMode batchMetricMode;
 
     private LongCounter pipelineMetricBatchCount;
@@ -50,8 +55,12 @@ public void updateBatchMetrics(QueueBatch batch) {
 
     private void updateBatchSizeMetric(QueueBatch batch) {
         long totalSize = 0L;
-        for (JrubyEventExtLibrary.RubyEvent rubyEvent : batch.events()) {
-            totalSize += rubyEvent.getEvent().estimateMemory();
+        try {
+            for (JrubyEventExtLibrary.RubyEvent rubyEvent : batch.events()) {
+                totalSize += rubyEvent.getEvent().estimateMemory();
+            }
+        } catch (IllegalArgumentException e) {
+            LOG.error("Failed to calculate batch byte size for metrics", e);
         }
         pipelineMetricBatchByteSize.increment(totalSize);
     }