Do not send headers to old kafka servers
Kafka message bundles with versions below 2 (e.d. 0.10) do not support
headers so do not inject them. Otherwise client gets really upset.
This is how similar check is being done in Kafka client itself:
@@ -17,9 +17,11 @@ import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.description.type.TypeDescription;
import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.record.RecordBatch;
@AutoService(Instrumenter.class)
public final class KafkaProducerInstrumentation extends Instrumenter.Default {
@@ -61,6 +63,7 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
@Advice.OnMethodEnter(suppress = Throwable.class)
public static Scope startSpan(
+ @Advice.FieldValue("apiVersions") final ApiVersions apiVersions,
@Advice.Argument(value = 0, readOnly = false) ProducerRecord record,
@Advice.Argument(value = 1, readOnly = false) Callback callback) {
final Scope scope = GlobalTracer.get().buildSpan("kafka.produce").startActive(false);
@@ -69,6 +72,8 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
callback = new ProducerCallback(callback, scope);
+ // Do not inject headers for batch versions below 2
+ if (apiVersions.maxUsableProduceMagic() >= RecordBatch.MAGIC_VALUE_V2) {
try {
GlobalTracer.get()
.inject(
@@ -92,6 +97,7 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
Format.Builtin.TEXT_MAP,
new TextMapInjectAdapter(record.headers()));
}
+ }
return scope;
}
