Support kafka streams 3 #4236
Support kafka streams 3 #4236
Conversation
laurit
requested review from
anuraaga,
iNikem,
jkwatson,
mateuszrzeszutek,
pavolloffay,
trask and
tylerbenson
as code owners
| public static <K, V> Iterator<ConsumerRecord<K, V>> wrap( | ||
| Iterator<ConsumerRecord<K, V>> delegateIterator, @Nullable SpanContext receiveSpanContext) { | ||
| if (KafkaTracingWrapperUtil.wrappingEnabled()) { | ||
| return new TracingIterator<>(delegateIterator, receiveSpanContext); | ||
| } | ||
| return delegateIterator; | ||
| } | ||
|
|
There was a problem hiding this comment.
I think there's a similar use case that @mateuszrzeszutek recently ran into recently:
would it work to use the same "workaround" for both? (either thread local suppression, or force unwrapping for both)
There was a problem hiding this comment.
I think I can use thread local for both, I'll just have to move the thread local to a different method that exists in all kafka-streams versions
There was a problem hiding this comment.
Let's wait until #4065 is merged though (I think it should be this week)
|
|
||
| package io.opentelemetry.javaagent.instrumentation.kafka; | ||
|
|
||
| public final class KafkaTracingWrapperUtil { |
There was a problem hiding this comment.
I wonder if we could rename it to sth like KafkaClientsConsumerProcessTracing - since it only affects the CONSUMER operation = "process" part of kafka-clients instrumentation
|
@laurit fyi (so we don't duplicate) I'm resolving merge conflicts |
|
@trask too late. I think that it might be better to put |
|
No worries, feel free to force push over my changes |
|
|
||
| package io.opentelemetry.javaagent.bootstrap.kafka; | ||
|
|
||
| public final class KafkaClientsConsumerProcessTracing { |
There was a problem hiding this comment.
can you add a brief comment about why this lives in the bootstrap class loader?
There was a problem hiding this comment.
added a comment, could you check whether it makes sense
|
|
||
| package io.opentelemetry.javaagent.bootstrap.kafka; | ||
|
|
||
| public interface KafkaClientsConsumerProcessWrapper<T> { |
| * @param compileString An extra dependency in the gradle canonical form: | ||
| * '<group_id>:<artifact_id>:<version_id>'. | ||
| * @param dependencyNotation An extra dependency in the gradle canonical form: | ||
| * '<group_id>:<artifact_id>:<version_id>' or a project dependency project(':some-project'). |
…to ensure that everybody uses the same copy of them
Resolves #4174
Apparently tests didn't pass with kafka-streams 1.x perhaps we should also test some intermediate versions?