avoid calling missing method on kafka-clients 4.x (#4136)

Author: SylvainJuge

CHANGELOG.next-release.md

@@ -10,6 +10,7 @@ This file contains all changes which are not released yet.
 # Fixes
 <!--FIXES-START-->
 * Prevent potential memory pressure by limiting OpenTelemetry metrics bridge attribute cache sizes - [#4123](https://github.com/elastic/apm-agent-java/pull/4123)
+* Fix `NoSuchMethodError` for Kafka 4136clients - [#4136](https://github.com/elastic/apm-agent-java/pull/4136)
 <!--FIXES-END-->
 # Features and enhancements
 <!--ENHANCEMENTS-START-->

apm-agent-plugins/apm-kafka-plugin/apm-kafka-headers-plugin/src/main/java/co/elastic/apm/agent/kafka/KafkaProducerHeadersInstrumentation.java

@@ -37,6 +37,10 @@
 
 import javax.annotation.Nullable;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+
 import static net.bytebuddy.implementation.bytecode.assign.Assigner.Typing.DYNAMIC;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
@@ -63,7 +67,8 @@ public String getAdviceClassName() {
     public static class KafkaProducerHeadersAdvice {
         private static final KafkaInstrumentationHelper helper = KafkaInstrumentationHelper.get();
         private static final KafkaInstrumentationHeadersHelper headersHelper = KafkaInstrumentationHeadersHelper.get();
-        private static boolean headersSupported = true;
+        @Nullable
+        private static Boolean headersSupported = null;
 
         @Nullable
         @Advice.AssignReturned.ToArguments(@ToArgument(value = 1, index = 1, typing = DYNAMIC))
@@ -73,9 +78,24 @@ public static Object[] beforeSend(@Advice.FieldValue("apiVersions") final ApiVer
                                           @Nullable @Advice.Argument(value = 1) Callback callback) {
             Span<?> span = helper.onSendStart(record);
 
-            // Avoid adding headers to records sent to a version older than 0.11.0 - see specifications in
-            // https://kafka.apache.org/0110/documentation.html#messageformat
-            if (apiVersions.maxUsableProduceMagic() >= RecordBatch.MAGIC_VALUE_V2 && headersSupported) {
+            if(headersSupported == null) {
+                try {
+                    // using method handle because method is gone in 4.0.0
+                    MethodHandle maxUsableProduceMagic = MethodHandles.lookup()
+                        .findVirtual(ApiVersions.class, "maxUsableProduceMagic", MethodType.methodType(byte.class));
+
+                    // Avoid adding headers to records sent to a version older than 0.11.0 - see specifications in
+                    // https://kafka.apache.org/0110/documentation.html#messageformat
+                    byte result = (byte)maxUsableProduceMagic.invoke(apiVersions);
+                    headersSupported = result >= RecordBatch.MAGIC_VALUE_V2;
+
+                } catch (Throwable e) {
+                    // method not present, assume 4.0.0 or later
+                    headersSupported = true;
+                }
+            }
+
+            if (headersSupported) {
                 try {
                     headersHelper.setOutgoingTraceContextHeaders(tracer.currentContext(), record);
                 } catch (final IllegalStateException e) {

apm-agent-plugins/apm-kafka-plugin/apm-kafka-headers-plugin/src/test/java/co/elastic/apm/agent/kafka/KafkaIT.java

@@ -59,6 +59,7 @@
 
 import javax.annotation.Nullable;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -384,8 +385,7 @@ private List<ConsumerRecord<String, String>> awaitReplyRecords(long timeoutMs, i
         long start = System.currentTimeMillis();
         long pollTime = 100;
         while (System.currentTimeMillis() + pollTime - start < timeoutMs) {
-            //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-            ConsumerRecords<String, String> records = replyConsumer.poll(pollTime);
+            ConsumerRecords<String, String> records = replyConsumer.poll(Duration.ofMillis(pollTime));
             if (!records.isEmpty()) {
                 records.forEach(replies::add);
             }
@@ -537,8 +537,7 @@ public void run() {
             kafkaConsumer.subscribe(Collections.singletonList(REQUEST_TOPIC));
             while (running) {
                 try {
-                    //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-                    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
+                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                     if (records != null && !records.isEmpty()) {
                         // Can't use switch because we run this test in a dedicated class loader, where the anonymous
                         // class created by the enum switch cannot be loaded

apm-agent-plugins/apm-kafka-plugin/apm-kafka-headers-plugin/src/test/java/co/elastic/apm/agent/kafka/KafkaIT_RealReporter.java

@@ -55,6 +55,7 @@
 import org.testcontainers.utility.DockerImageName;
 
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
@@ -301,8 +302,7 @@ private List<ConsumerRecord<String, String>> awaitReplyRecords(long timeoutMs, i
         long start = System.currentTimeMillis();
         long pollTime = 100;
         while (System.currentTimeMillis() + pollTime - start < timeoutMs) {
-            //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-            ConsumerRecords<String, String> records = replyConsumer.poll(pollTime);
+            ConsumerRecords<String, String> records = replyConsumer.poll(Duration.ofMillis(pollTime));
             if (!records.isEmpty()) {
                 records.forEach(replies::add);
             }
@@ -350,8 +350,7 @@ public void run() {
             kafkaConsumer.subscribe(Collections.singletonList(REQUEST_TOPIC));
             while (running) {
                 try {
-                    //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-                    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
+                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                     if (records != null && !records.isEmpty()) {
                         // Can't use switch because we run this test in a dedicated class loader, where the anonymous
                         // class created by the enum switch cannot be loaded

apm-agent-plugins/apm-kafka-plugin/apm-kafka-headers-plugin/src/test/java/co/elastic/apm/agent/kafka/KafkaLegacyBrokerIT.java

@@ -49,6 +49,7 @@
 import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
 import javax.annotation.Nullable;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
@@ -275,8 +276,7 @@ private void sendTwoRecordsAndConsumeReplies() {
             int expectedSpans = (testScenario == TestScenario.NO_CONTEXT_PROPAGATION) ? 2 : 4;
             await().atMost(500, MILLISECONDS).until(() -> reporter.getSpans().size() == expectedSpans);
         }
-        //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-        ConsumerRecords<String, String> replies = replyConsumer.poll(2000);
+        ConsumerRecords<String, String> replies = replyConsumer.poll(Duration.ofMillis(2000));
         assertThat(callback).isNotEmpty();
         assertThat(replies.count()).isEqualTo(2);
         Iterator<ConsumerRecord<String, String>> iterator = replies.iterator();
@@ -395,8 +395,7 @@ public void run() {
             kafkaConsumer.subscribe(Collections.singletonList(REQUEST_TOPIC));
             while (running) {
                 try {
-                    //noinspection deprecation - this poll overload is deprecated in newer clients, but enables testing of old ones
-                    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
+                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                     if (records != null && !records.isEmpty()) {
                         // Can't use switch because we run this test in a dedicated class loader, where the anonymous
                         // class created by the enum switch cannot be loaded

apm-agent-plugins/apm-opentelemetry/apm-opentelemetry-metrics-bridge-parent/apm-opentelemetry-metrics-bridge-v1_14/src/test/java/co/elastic/apm/agent/opentelemetry/metrics/bridge/BridgeFactoryV1_14Test.java

@@ -1,3 +1,21 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
 package co.elastic.apm.agent.opentelemetry.metrics.bridge;
 
 import io.opentelemetry.api.common.AttributeKey;