[BUG] ParseJson processor overwrites fields in Otel Log Event causing pipeline to shut down

[oeyh]

Describe the bug
In a pipeline with OTEL log source and parse_json processor, it's possible that the processor overwrites attributes field in the event, changing its value from a map to a string. This will subsequently cause ClassCastException in the sink when it calls JacksonOtelLog.toJsonString(). The exception is currently not caught and will trigger a pipeline shutdown.

To Reproduce
With this pipeline config:

version: "2"
otel-logs-pipeline:
  source:
    otel_logs_source:
      path: "/${pipelineName}/v1/logs"
      unframed_requests: true
      ssl: false
  processor:
    - parse_json:
        source: "body"
  sink:
    - stdout:

and this test data:

{
  "resourceLogs": [
    {
      "resource": {
        "attributes": [
          {
            "key": "resource-attr",
            "value": {
              "stringValue": "resource-attr-val-01"
            }
          }
        ]
      },
      "scopeLogs": [
        {
          "scope": {},
          "logRecords": [
            {

              "name": "logA",
              "body": {
                "stringValue": "{\"attributes\": \"hello\"}"
              },
              "attributes": [
                {
                  "key": "app",
                  "value": {
                    "stringValue": "server"
                  }
                },
                {
                  "key": "instance_num",
                  "value": {
                    "intValue": "1"
                  }
                }
              ]
            }
          ]
        }
      ]
    }
  ]
}

Pipeline shutdown will be triggered:

2023-08-16T23:38:15,731 [otel-logs-pipeline-sink-worker-2-thread-1] ERROR org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor - Pipeline [otel-logs-pipeline] process worker encountered a fatal exception, cannot proceed further
java.util.concurrent.ExecutionException: java.lang.ClassCastException: class com.fasterxml.jackson.databind.node.TextNode cannot be cast to class com.fasterxml.jackson.databind.node.ObjectNode (com.fasterxml.jackson.databind.node.TextNode and com.fasterxml.jackson.databind.node.ObjectNode are in unnamed module of loader 'app')
	at java.util.concurrent.FutureTask.report(FutureTask.java:122) ~[?:?]
	at java.util.concurrent.FutureTask.get(FutureTask.java:191) ~[?:?]
	at org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor.afterExecute(PipelineThreadPoolExecutor.java:70) [data-prepper-core-2.4.0-SNAPSHOT.jar:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: java.lang.ClassCastException: class com.fasterxml.jackson.databind.node.TextNode cannot be cast to class com.fasterxml.jackson.databind.node.ObjectNode (com.fasterxml.jackson.databind.node.TextNode and com.fasterxml.jackson.databind.node.ObjectNode are in unnamed module of loader 'app')
	at org.opensearch.dataprepper.model.log.JacksonOtelLog.toJsonString(JacksonOtelLog.java:112) ~[data-prepper-api-2.4.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.model.event.JacksonEvent$JsonStringBuilder.toJsonString(JacksonEvent.java:578) ~[data-prepper-api-2.4.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.plugins.sink.StdOutSink.checkTypeAndPrintObject(StdOutSink.java:56) ~[common-2.4.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.plugins.sink.StdOutSink.output(StdOutSink.java:48) ~[common-2.4.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.lambda$publishToSinks$5(Pipeline.java:336) ~[data-prepper-core-2.4.0-SNAPSHOT.jar:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	... 2 more

Expected behavior
Pipeline should continue to run in the above situation.

Screenshots
If applicable, add screenshots to help explain your problem.

Environment (please complete the following information):

• OS: [e.g. Ubuntu 20.04 LTS]
• Version [e.g. 22]

Additional context
Add any other context about the problem here.

[oeyh]

There are two issues causing the shutdown:

1. parse_json processor can overwrite fields in Otel log event. We can probably add an overwrite_if_key_exists option similar to add_entries processor and defaults it to false.
2. ClassCastException in this line of code in JacksonOtelLog.toJsonString() should be caught and handled to avoid pipeline shutdown.