mutable segment indexing exception log floods and cause high cpu issue.

[wirybeaver]

Check the function call stack below. If the every row of input data is malformed, the consumeLoop will always exit with exception and java thread will print the stack, which will cause high cpu. Shall we log the exception with a sampling rate for comsuming Loop? We also need a metrics to report the exception times during consumption. cc: @ankitsultana

{"@timestamp":"2024-10-23T22:03:02.376+00:00","@Version":1,"message":"failed to index value with json_index","logger_name":"org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl_<private_info>","thread_name":"<private_info>","level":"ERROR","level_value":40000,"stack_trace":"java.lang.IllegalArgumentException: Caught exception while flattening JSON node: with config: {"disabled":false,"maxLevels":-1,"excludeArray":false,"disableCrossArrayUnnest":false,"maxValueLength":0,"skipInvalidJson":true}\n\tat org.apache.pinot.spi.utils.JsonUtils.flatten(JsonUtils.java:370)\n\tat org.apache.pinot.spi.utils.JsonUtils.flatten(JsonUtils.java:737)\n\tat org.apache.pinot.segment.local.realtime.impl.json.MutableJsonIndexImpl.add(MutableJsonIndexImpl.java:82)\n\tat org.apache.pinot.segment.spi.index.mutable.MutableJsonIndex.add(MutableJsonIndex.java:32)\n\tat org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl.addNewRow(MutableSegmentImpl.java:705)\n\tat org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl.index(MutableSegmentImpl.java:531)\n\tat org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.processStreamEvents(RealtimeSegmentDataManager.java:607)\n\tat org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.consumeLoop(RealtimeSegmentDataManager.java:457)\n\tat org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager$PartitionConsumer.run(RealtimeSegmentDataManager.java:695)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.lang.IllegalArgumentException: Unexpected node type: MISSING\n\tat fievel.shaded.streaming.pinot.com.google.common.base.Preconditions.checkArgument(Preconditions.java:218)\n\tat org.apache.pinot.spi.utils.JsonUtils.flatten(JsonUtils.java:391)\n\tat org.apache.pinot.spi.utils.JsonUtils.flatten(JsonUtils.java:363)\n\t... 9 common frames omitted\n",<private_info>}