NPE when KafkaConsumerRecordGetter call getAll #14329
Description
otel version: 2.16.0
KafkaConsumerRecordGetter getAll causes NPE if kafka headers have null value.
Here is my stack trace
java.lang.NullPointerException: Cannot read the array length because "bytes" is null
at java.base/java.lang.String.<init>(String.java:1425)
at io.opentelemetry.javaagent.shaded.instrumentation.kafkaclients.common.v0_11.internal.KafkaConsumerRecordGetter.lambda$getAll$0(KafkaConsumerRecordGetter.java:43)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.Spliterators$IteratorSpliterator.tryAdvance(Spliterators.java:1950)
at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.lambda$initPartialTraversalState$0(StreamSpliterators.java:292)
at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.fillBuffer(StreamSpliterators.java:206)
at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.doAdvance(StreamSpliterators.java:161)
at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.tryAdvance(StreamSpliterators.java:298)
at java.base/java.util.Spliterators$1Adapter.hasNext(Spliterators.java:681)
at io.opentelemetry.javaagent.shaded.io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator.extractMulti(W3CBaggagePropagator.java:112)
at io.opentelemetry.javaagent.shaded.io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator.extract(W3CBaggagePropagator.java:99)
at io.opentelemetry.javaagent.shaded.io.opentelemetry.context.propagation.MultiTextMapPropagator.extract(MultiTextMapPropagator.java:65)
at io.opentelemetry.javaagent.shaded.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor.extract(PropagatorBasedSpanLinksExtractor.java:33)
at io.opentelemetry.javaagent.shaded.instrumentation.api.instrumenter.Instrumenter.doStartImpl(Instrumenter.java:186)
at io.opentelemetry.javaagent.shaded.instrumentation.api.instrumenter.Instrumenter.doStart(Instrumenter.java:169)
at io.opentelemetry.javaagent.shaded.instrumentation.api.instrumenter.Instrumenter.start(Instrumenter.java:136)
at io.opentelemetry.javaagent.shaded.instrumentation.kafkaclients.common.v0_11.internal.TracingIterator.next(TracingIterator.java:82)
at io.opentelemetry.javaagent.shaded.instrumentation.kafkaclients.common.v0_11.internal.TracingIterator.next(TracingIterator.java:20)
Suggestion:
@Override
public Iterator<String> getAll(@Nullable KafkaProcessRequest carrier, String key) {
return StreamSupport.stream(carrier.getRecord().headers().headers(key).spliterator(), false)
.map(Header::value)
.filter(Objects::nonNull)
.map(value -> new String(value, StandardCharsets.UTF_8))
.iterator();
}
}