NPE when KafkaConsumerRecordGetter call getAll

[charlie-world]

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();
  }
}