fix muzzle for kafka

nayeem-kamal · nayeem-kamal · commit 50f341ab0ff2 · 2024-11-07T16:56:13.000-05:00
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/build.gradle b/dd-java-agent/instrumentation/kafka-clients-3.8/build.gradle
@@ -6,7 +6,7 @@ muzzle {
     group = "org.apache.kafka"
     module = "kafka-clients"
     versions = "[3.8.0,)"
-    assertInverse = false
+    assertInverse = true
   }
 }
 
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConstructorAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConstructorAdvice.java
@@ -7,7 +7,6 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;
 import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
@@ -54,9 +53,8 @@ public static void captureGroup(
     }
   }
 
-  public static void muzzleCheck(ConsumerRecord record) {
-    // KafkaConsumerInstrumentation only applies for kafka versions with headers
-    // Make an explicit call so KafkaConsumerGroupInstrumentation does the same
-    record.headers();
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
   }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorAdvice.java
@@ -12,9 +12,9 @@
 import java.util.Map;
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.Metadata;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 import org.apache.kafka.clients.consumer.internals.RequestFuture;
 import org.apache.kafka.common.TopicPartition;
 
@@ -66,9 +66,8 @@ public static void trackCommitOffset(
     }
   }
 
-  public static void muzzleCheck(ConsumerRecord record) {
-    // KafkaConsumerInstrumentation only applies for kafka versions with headers
-    // Make an explicit call so ConsumerCoordinatorInstrumentation does the same
-    record.headers();
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
   }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java
@@ -5,6 +5,7 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class IterableAdvice {
 
@@ -31,4 +32,9 @@ public static void wrap(
               bootstrapServers);
     }
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java
@@ -6,6 +6,7 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class IteratorAdvice {
 
@@ -32,4 +33,9 @@ public static void wrap(
               bootstrapServers);
     }
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java
@@ -3,6 +3,7 @@
 import datadog.trace.api.Config;
 import datadog.trace.bootstrap.ContextStore;
 import org.apache.kafka.clients.Metadata;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class KafkaConsumerInstrumentationHelper {
   public static String extractGroup(KafkaConsumerInfo kafkaConsumerInfo) {
@@ -26,4 +27,9 @@ public static String extractClusterId(
   public static String extractBootstrapServers(KafkaConsumerInfo kafkaConsumerInfo) {
     return kafkaConsumerInfo == null ? null : kafkaConsumerInfo.getBootstrapServers().get();
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/LegacyConstructorAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/LegacyConstructorAdvice.java
@@ -6,9 +6,9 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;
 import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class LegacyConstructorAdvice {
   // new - capture the ConsumerDelegate instead of KafkaConsumer
@@ -46,9 +46,8 @@ public static void captureGroup(
     }
   }
 
-  public static void muzzleCheck(ConsumerRecord record) {
-    // KafkaConsumerInstrumentation only applies for kafka versions with headers
-    // Make an explicit call so KafkaConsumerGroupInstrumentation does the same
-    record.headers();
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
   }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java
@@ -9,6 +9,7 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class ListAdvice {
 
@@ -30,4 +31,9 @@ public static void wrap(
               iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers);
     }
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/MetadataUpdate22AndAfterAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/MetadataUpdate22AndAfterAdvice.java
@@ -3,7 +3,7 @@
 import datadog.trace.bootstrap.InstrumentationContext;
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.Metadata;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 import org.apache.kafka.common.requests.MetadataResponse;
 
 public class MetadataUpdate22AndAfterAdvice {
@@ -15,9 +15,8 @@ public static void onEnter(
     }
   }
 
-  public static void muzzleCheck(ConsumerRecord record) {
-    // KafkaConsumerInstrumentation only applies for kafka versions with headers
-    // Make an explicit call so MetadataInstrumentation does the same
-    record.headers();
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
   }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/MetadataUpdateBefore22Advice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/MetadataUpdateBefore22Advice.java
@@ -3,7 +3,7 @@
 import datadog.trace.bootstrap.InstrumentationContext;
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.Metadata;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 import org.apache.kafka.common.Cluster;
 
 public class MetadataUpdateBefore22Advice {
@@ -16,9 +16,8 @@ public static void onEnter(
     }
   }
 
-  public static void muzzleCheck(ConsumerRecord record) {
-    // KafkaConsumerInstrumentation only applies for kafka versions with headers
-    // Make an explicit call so MetadataInstrumentation does the same
-    record.headers();
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
   }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/OffsetCommitCallbackInvokerAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/OffsetCommitCallbackInvokerAdvice.java
@@ -15,4 +15,9 @@ public static void onEnqueue(
             .get(callbackInvoker);
     callback = new DDOffsetCommitCallback(callback, kafkaConsumerInfo);
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/PayloadSizeAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/PayloadSizeAdvice.java
@@ -5,6 +5,7 @@
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
 import datadog.trace.bootstrap.instrumentation.api.StatsPoint;
 import net.bytebuddy.asm.Advice;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 public class PayloadSizeAdvice {
 
@@ -32,4 +33,9 @@ public static void onEnter(@Advice.Argument(value = 0) int estimatedPayloadSize)
       AgentTracer.get().getDataStreamsMonitoring().add(updated);
     }
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ProducerAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ProducerAdvice.java
@@ -23,6 +23,7 @@
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.Metadata;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -114,4 +115,9 @@ public static void stopSpan(
     PRODUCER_DECORATE.beforeFinish(scope);
     scope.close();
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java b/dd-java-agent/instrumentation/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java
@@ -13,6 +13,7 @@
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;
+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;
 
 /**
  * this method transfers the consumer group from the KafkaConsumer class key to the ConsumerRecords
@@ -60,4 +61,9 @@ public static void captureGroup(
     span.finish();
     scope.close();
   }
+
+  public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {
+    // Only applies for kafka versions with OffsetCommitCallbackInvoker
+    invoker.executeCallbacks();
+  }
 }

Original file line number	Diff line number	Diff line change
`@@ -6,7 +6,7 @@ muzzle {`
`6`	`6`	`group = "org.apache.kafka"`
`7`	`7`	`module = "kafka-clients"`
`8`	`8`	`versions = "[3.8.0,)"`
`9`		`- assertInverse = false`
	`9`	`+ assertInverse = true`
`10`	`10`	`}`
`11`	`11`	`}`
`12`	`12`
Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,6 @@`
`7`	`7`	`import org.apache.kafka.clients.Metadata;`
`8`	`8`	`import org.apache.kafka.clients.consumer.ConsumerConfig;`
`9`	`9`	`import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;`
`10`		`-import org.apache.kafka.clients.consumer.ConsumerRecord;`
`11`	`10`	`import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;`
`12`	`11`	`import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`13`	`12`
`@@ -54,9 +53,8 @@ public static void captureGroup(`
`54`	`53`	`}`
`55`	`54`	`}`
`56`	`55`
`57`		`- public static void muzzleCheck(ConsumerRecord record) {`
`58`		`- // KafkaConsumerInstrumentation only applies for kafka versions with headers`
`59`		`- // Make an explicit call so KafkaConsumerGroupInstrumentation does the same`
`60`		`- record.headers();`
	`56`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`57`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`58`	`+ invoker.executeCallbacks();`
`61`	`59`	`}`
`62`	`60`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,9 +12,9 @@`
`12`	`12`	`import java.util.Map;`
`13`	`13`	`import net.bytebuddy.asm.Advice;`
`14`	`14`	`import org.apache.kafka.clients.Metadata;`
`15`		`-import org.apache.kafka.clients.consumer.ConsumerRecord;`
`16`	`15`	`import org.apache.kafka.clients.consumer.OffsetAndMetadata;`
`17`	`16`	`import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;`
	`17`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`18`	`18`	`import org.apache.kafka.clients.consumer.internals.RequestFuture;`
`19`	`19`	`import org.apache.kafka.common.TopicPartition;`
`20`	`20`
`@@ -66,9 +66,8 @@ public static void trackCommitOffset(`
`66`	`66`	`}`
`67`	`67`	`}`
`68`	`68`
`69`		`- public static void muzzleCheck(ConsumerRecord record) {`
`70`		`- // KafkaConsumerInstrumentation only applies for kafka versions with headers`
`71`		`- // Make an explicit call so ConsumerCoordinatorInstrumentation does the same`
`72`		`- record.headers();`
	`69`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`70`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`71`	`+ invoker.executeCallbacks();`
`73`	`72`	`}`
`74`	`73`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`	`import org.apache.kafka.clients.Metadata;`
`6`	`6`	`import org.apache.kafka.clients.consumer.ConsumerRecord;`
`7`	`7`	`import org.apache.kafka.clients.consumer.ConsumerRecords;`
	`8`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`8`	`9`
`9`	`10`	`public class IterableAdvice {`
`10`	`11`
`@@ -31,4 +32,9 @@ public static void wrap(`
`31`	`32`	`bootstrapServers);`
`32`	`33`	`}`
`33`	`34`	`}`
	`35`	`+`
	`36`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`37`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`38`	`+ invoker.executeCallbacks();`
	`39`	`+ }`
`34`	`40`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@`
`6`	`6`	`import org.apache.kafka.clients.Metadata;`
`7`	`7`	`import org.apache.kafka.clients.consumer.ConsumerRecord;`
`8`	`8`	`import org.apache.kafka.clients.consumer.ConsumerRecords;`
	`9`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`9`	`10`
`10`	`11`	`public class IteratorAdvice {`
`11`	`12`
`@@ -32,4 +33,9 @@ public static void wrap(`
`32`	`33`	`bootstrapServers);`
`33`	`34`	`}`
`34`	`35`	`}`
	`36`	`+`
	`37`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`38`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`39`	`+ invoker.executeCallbacks();`
	`40`	`+ }`
`35`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,9 +6,9 @@`
`6`	`6`	`import org.apache.kafka.clients.Metadata;`
`7`	`7`	`import org.apache.kafka.clients.consumer.ConsumerConfig;`
`8`	`8`	`import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;`
`9`		`-import org.apache.kafka.clients.consumer.ConsumerRecord;`
`10`	`9`	`import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;`
`11`	`10`	`import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;`
	`11`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`12`	`12`
`13`	`13`	`public class LegacyConstructorAdvice {`
`14`	`14`	`// new - capture the ConsumerDelegate instead of KafkaConsumer`
`@@ -46,9 +46,8 @@ public static void captureGroup(`
`46`	`46`	`}`
`47`	`47`	`}`
`48`	`48`
`49`		`- public static void muzzleCheck(ConsumerRecord record) {`
`50`		`- // KafkaConsumerInstrumentation only applies for kafka versions with headers`
`51`		`- // Make an explicit call so KafkaConsumerGroupInstrumentation does the same`
`52`		`- record.headers();`
	`49`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`50`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`51`	`+ invoker.executeCallbacks();`
`53`	`52`	`}`
`54`	`53`	`}`
Original file line number	Diff line number	Diff line change
`@@ -9,6 +9,7 @@`
`9`	`9`	`import org.apache.kafka.clients.Metadata;`
`10`	`10`	`import org.apache.kafka.clients.consumer.ConsumerRecord;`
`11`	`11`	`import org.apache.kafka.clients.consumer.ConsumerRecords;`
	`12`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`12`	`13`
`13`	`14`	`public class ListAdvice {`
`14`	`15`
`@@ -30,4 +31,9 @@ public static void wrap(`
`30`	`31`	`iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers);`
`31`	`32`	`}`
`32`	`33`	`}`
	`34`	`+`
	`35`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`36`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`37`	`+ invoker.executeCallbacks();`
	`38`	`+ }`
`33`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@`
`3`	`3`	`import datadog.trace.bootstrap.InstrumentationContext;`
`4`	`4`	`import net.bytebuddy.asm.Advice;`
`5`	`5`	`import org.apache.kafka.clients.Metadata;`
`6`		`-import org.apache.kafka.clients.consumer.ConsumerRecord;`
	`6`	`+import org.apache.kafka.clients.consumer.internals.OffsetCommitCallbackInvoker;`
`7`	`7`	`import org.apache.kafka.common.requests.MetadataResponse;`
`8`	`8`
`9`	`9`	`public class MetadataUpdate22AndAfterAdvice {`
`@@ -15,9 +15,8 @@ public static void onEnter(`
`15`	`15`	`}`
`16`	`16`	`}`
`17`	`17`
`18`		`- public static void muzzleCheck(ConsumerRecord record) {`
`19`		`- // KafkaConsumerInstrumentation only applies for kafka versions with headers`
`20`		`- // Make an explicit call so MetadataInstrumentation does the same`
`21`		`- record.headers();`
	`18`	`+ public static void muzzleCheck(OffsetCommitCallbackInvoker invoker) {`
	`19`	`+ // Only applies for kafka versions with OffsetCommitCallbackInvoker`
	`20`	`+ invoker.executeCallbacks();`
`22`	`21`	`}`
`23`	`22`	`}`