Please help me with the below issue

Data Prepper
,
1

Hi all, I’ve been working on creating an application to collect user activity in the form of traces using opentelemetry sdk for browser and django. I am trying to send the traces to opensearch via this route - collector → aws msk (kafka) → opensearch ingestion pipeline (data-prepper) → opensearch dashboards
We are facing an issue in trace format in the above case whereas sending directly to opensearch ingestion pipeline from collector, it is able to process it correctly. On closer look i found this -

**Using kafka -**

"resourceSpans": [
      {
        "resource": {
          "attributes": [
            {
              "value": {
                "stringValue": "backend"
              },
              "key": "service.name"
            }
          ]
        },
        "scopeSpans": [
          {
            "spans": [
              {
                "traceId": "3a50ceea8f1c5e8a1ea5a31738e0bcb1",
                "spanId": "b82407ae305977ae",
                "endTimeUnixNano": "1722319922064817700",
                "kind": 2,
                "flags": 256,
                "name": "GET ^.*",
                "attributes": [
                  {
                    "value": {
                      "arrayValue": {
                        "values": [
                          {
                            "stringValue": "http://44.200.10.149:4318/v1/traces"
                          }
                        ]
                      }
                    },
                    "key": "http.request.header.trace_endpoint"
                  },
                  {
                    "value": {
                      "arrayValue": {
                        "values": [
                          {
                            "stringValue": "127.0.0.1:8000"
                          }
                        ]
                      }
                    },
                    "key": "http.request.header.host"
                  },
                  {
                    "value": {
                      "arrayValue": {
                        "values": [
                          {
                            "stringValue": "keep-alive"
                          }
                        ]
                      }
                    } ........... ```

**without kafka (this one is working with opensearch) -**

   
 "traceId": "9ae50ff1c0c84da8ba0ec34f1827bb55",
    "droppedLinksCount": 0,
    "kind": "SPAN_KIND_SERVER",
    "droppedEventsCount": 0,
    "traceGroupFields": {
      "endTime": "2024-08-29T13:21:53.441322500Z",
      "durationInNanos": 3554749800,
      "statusCode": 0
    },
    "traceGroup": "POST api/receive-traces/",
    "serviceName": "backend",
    "parentSpanId": "",
    "spanId": "c61bb3bcf3f8e439",
    "traceState": "",
    "name": "POST api/receive-traces/",
    "startTime": "2024-08-29T13:21:49.886572700Z",
    "links": [],
    "endTime": "2024-08-29T13:21:53.441322500Z",
    "droppedAttributesCount": 0,
    "durationInNanos": 3554749800,
    "events": [],

How can i use kafka and have same format for traces ???