Added parsing otel json to span from kafka topic to Span by Mamol27 · Pull Request #5681 · opensearch-project/data-prepper

Mamol27 · 2025-05-06T15:18:41Z

Description

Parsing of OpenTelemetry trace in JSON format into Span has been implemented to support trace handling via Kafka.
We had to implement it, as the client requires all data to be sent exclusively through Kafka.
To implement this, a new MessageFormat was added.

Sorry for creating new PR I've got a problem with solving DCO. (Old PR #5590)

Issues Resolved

Resolves #5446

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: mamol27 <mamol27@yandex.ru>

KarstenSchnitter

Thanks for providing this PR. I did not review the tests. But I have some comments on the main implementation.

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

KarstenSchnitter · 2025-05-06T18:23:05Z

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

+                    final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, Instant.now().toEpochMilli());
+
+                    KafkaOtelJsonTraceParser kafkaOtelJsonTraceParser = new KafkaOtelJsonTraceParser();
+                    kafkaOtelJsonTraceParser.parse(consumerRecord.value().toString(), Instant.ofEpochMilli(receivedTimeStamp), (record) -> {


@kkondaka: I think this implementation could be improved, if the type of consumerRecord.value() was known. If it was byte[], ByteBuffer or Bytes, it could be wrapped in an java.io.Reader implementation that avoids generating an intermediate string. The JsonFormat.parser().merge(...) would be fine with such a reader. Unfortunately, if I understand the code correctly, the type T is a free placeholder. Unless any valid assumptions on the type can be made, this improvement is not possible.

Using an InputCodec as I suggested elsewhere would allow us to read from an InputStream. So you could have new ByteArrayInputStream(consumerRecord.value()).

consumerRecord.value() isn't byte[] it has type T

.../src/main/java/org/opensearch/dataprepper/plugins/kafka/parser/KafkaOtelJsonTraceParser.java

dlvenable

Thank you @Mamol27 for this great contribution!

.../src/main/java/org/opensearch/dataprepper/plugins/kafka/parser/KafkaOtelJsonTraceParser.java

...kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/MessageFormat.java

dlvenable · 2025-05-06T20:26:35Z

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

                        processRecord(acknowledgementSet, record);
                    }
+                } else if (schema == MessageFormat.JSON_OTEL_TRACE) {
+                    final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, Instant.now().toEpochMilli());


Why is this code not common to all reads?

because it process many records instead one to other

dlvenable · 2025-05-06T20:29:50Z

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

+                    final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, Instant.now().toEpochMilli());
+
+                    KafkaOtelJsonTraceParser kafkaOtelJsonTraceParser = new KafkaOtelJsonTraceParser();
+                    kafkaOtelJsonTraceParser.parse(consumerRecord.value().toString(), Instant.ofEpochMilli(receivedTimeStamp), (record) -> {


Using an InputCodec as I suggested elsewhere would allow us to read from an InputStream. So you could have new ByteArrayInputStream(consumerRecord.value()).

Signed-off-by: mamol27 <mamol27@yandex.ru>

Parsing with TracesData.Builder have some issuies Signed-off-by: mamol27 <mamol27@yandex.ru>

Mamol27 · 2025-05-14T13:07:14Z

@dlvenable @KarstenSchnitter

All IT tests passed locally, is that GitHub issue? Could you please give me some advice?

Signed-off-by: mamol27 <mamol27@yandex.ru>

# Conflicts: # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerFactory.java

Signed-off-by: mamol27 <mamol27@yandex.ru>

...src/test/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerTest.java

...-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelTraceInputCodec.java

...n/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelTraceInputCodecConfig.java

...common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OtelTraceJsonDecoder.java

kkondaka · 2025-12-08T23:56:26Z

@Mamol27 sorry for the delay in reviewing this. Please address the comments.

Signed-off-by: mamol27 <mamol27@yandex.ru>

Added parsing otel json to span from kafka topic to Span

f1c60a0

Signed-off-by: mamol27 <mamol27@yandex.ru>

Mamol27 requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners May 6, 2025 15:18

KarstenSchnitter reviewed May 6, 2025

View reviewed changes

dlvenable requested changes May 6, 2025

View reviewed changes

json-to-trace codec moved to otel-proto-common

e5494f3

Signed-off-by: mamol27 <mamol27@yandex.ru>

Mamol27 requested review from KarstenSchnitter and dlvenable May 13, 2025 08:40

Mamol27 added 2 commits May 14, 2025 11:01

made codec for KafkaCustomConsumer

226d7fa

Parsing with TracesData.Builder have some issuies Signed-off-by: mamol27 <mamol27@yandex.ru>

made codec for KafkaCustomConsumer

8969ccc

Parsing with TracesData.Builder have some issuies Signed-off-by: mamol27 <mamol27@yandex.ru>

Mamol27 added 4 commits June 4, 2025 10:57

fixed issue with null value in attribute and long value in intValue

134fee2

Signed-off-by: mamol27 <mamol27@yandex.ru>

fixed issue with null value in attribute and long value in intValue

bcdfe24

Signed-off-by: mamol27 <mamol27@yandex.ru>

resolve merge conflict

02959e1

Signed-off-by: mamol27 <mamol27@yandex.ru>

kkondaka requested changes Dec 8, 2025

View reviewed changes

comments corrected

ce19a42

Signed-off-by: mamol27 <mamol27@yandex.ru>

Mamol27 requested a review from kkondaka December 11, 2025 07:37

Conversation

Mamol27 commented May 6, 2025

Description

Issues Resolved

Check List

Uh oh!

KarstenSchnitter left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

KarstenSchnitter May 6, 2025

Choose a reason for hiding this comment

Uh oh!

dlvenable May 6, 2025

Choose a reason for hiding this comment

Uh oh!

Mamol27 May 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dlvenable May 6, 2025

Choose a reason for hiding this comment

Uh oh!

Mamol27 May 7, 2025

Choose a reason for hiding this comment

Uh oh!

dlvenable May 6, 2025

Choose a reason for hiding this comment

Uh oh!

Mamol27 commented May 14, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kkondaka commented Dec 8, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Mamol27 May 7, 2025 •

edited

Loading