avoid calling missing method on kafka-clients 4.x

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;