dougqh · Sep 28, 2026
@@ -6,6 +6,7 @@
 import static datadog.trace.instrumentation.kafka_clients.KafkaDecorator.KAFKA_PRODUCED_KEY;
 
 import datadog.trace.api.Config;
+import datadog.trace.api.Functions;
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation;
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation.ContextVisitor;
 import java.nio.ByteBuffer;
@@ -20,14 +21,21 @@ public class TextMapExtractAdapter implements ContextVisitor<Headers> {
   private static final Logger log = LoggerFactory.getLogger(TextMapExtractAdapter.class);
 
   public static final TextMapExtractAdapter GETTER =
-      new TextMapExtractAdapter(Config.get().isKafkaClientBase64DecodingEnabled());
+      new TextMapExtractAdapter(
+          Config.get().isKafkaClientBase64DecodingEnabled(),
+          Config.get().isKafkaClientBase64DecodingGuardEnabled());
 
   private final Function<byte[], String> headerValueTransformer;
   private final Base64.Decoder decoder;
 
   public TextMapExtractAdapter(boolean decodeBase64Headers) {
+    this(decodeBase64Headers, true);
+  }
+
+  public TextMapExtractAdapter(boolean decodeBase64Headers, boolean guardEnabled) {
     if (decodeBase64Headers) {
-      this.headerValueTransformer = BASE64_DECODE;
+      this.headerValueTransformer =
+          guardEnabled ? new Functions.GuardedBase64Decode()::decodeOrNull : BASE64_DECODE;
       this.decoder = Base64.getDecoder();
     } else {
       this.headerValueTransformer = UTF8_BYTES_TO_STRING;

@@ -5,6 +5,7 @@
 import static datadog.trace.api.telemetry.LogCollector.EXCLUDE_TELEMETRY;
 
 import datadog.trace.api.Config;
+import datadog.trace.api.Functions;
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation;
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation.ContextVisitor;
 import java.nio.ByteBuffer;
@@ -19,14 +20,21 @@ public class TextMapExtractAdapter implements ContextVisitor<Headers> {
   private static final Logger log = LoggerFactory.getLogger(TextMapExtractAdapter.class);
 
   public static final TextMapExtractAdapter GETTER =
-      new TextMapExtractAdapter(Config.get().isKafkaClientBase64DecodingEnabled());
+      new TextMapExtractAdapter(
+          Config.get().isKafkaClientBase64DecodingEnabled(),
+          Config.get().isKafkaClientBase64DecodingGuardEnabled());
 
   private final Function<byte[], String> headerValueTransformer;
   private final Base64.Decoder decoder;
 
   public TextMapExtractAdapter(boolean decodeBase64Headers) {
+    this(decodeBase64Headers, true);
+  }
+
+  public TextMapExtractAdapter(boolean decodeBase64Headers, boolean guardEnabled) {
     if (decodeBase64Headers) {
-      this.headerValueTransformer = BASE64_DECODE;
+      this.headerValueTransformer =
+          guardEnabled ? new Functions.GuardedBase64Decode()::decodeOrNull : BASE64_DECODE;
       this.decoder = Base64.getDecoder();
     } else {
       this.headerValueTransformer = UTF8_BYTES_TO_STRING;

@@ -111,6 +111,8 @@ public final class TraceInstrumentationConfig {
       "kafka.client.propagation.disabled.topics";
   public static final String KAFKA_CLIENT_BASE64_DECODING_ENABLED =
       "kafka.client.base64.decoding.enabled";
+  public static final String KAFKA_CLIENT_BASE64_DECODING_GUARD_ENABLED =
+      "kafka.client.base64.decoding.guard.enabled";
 
   public static final String JMS_PROPAGATION_DISABLED_TOPICS = "jms.propagation.disabled.topics";
   public static final String JMS_PROPAGATION_DISABLED_QUEUES = "jms.propagation.disabled.queues";

@@ -616,6 +616,7 @@
 import static datadog.trace.api.config.TraceInstrumentationConfig.JMS_PROPAGATION_DISABLED_TOPICS;
 import static datadog.trace.api.config.TraceInstrumentationConfig.JMS_UNACKNOWLEDGED_MAX_AGE;
 import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_BASE64_DECODING_ENABLED;
+import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_BASE64_DECODING_GUARD_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS;
 import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION;
 import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION_ENABLED;
@@ -1304,6 +1305,7 @@ public static String getHostName() {
   private final boolean kafkaClientPropagationEnabled;
   private final Set<String> kafkaClientPropagationDisabledTopics;
   private final boolean kafkaClientBase64DecodingEnabled;
+  private final boolean kafkaClientBase64DecodingGuardEnabled;
 
   private final boolean jmsPropagationEnabled;
   private final Set<String> jmsPropagationDisabledTopics;
@@ -3117,6 +3119,8 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment())
         tryMakeImmutableSet(configProvider.getList(KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS));
     kafkaClientBase64DecodingEnabled =
         configProvider.getBoolean(KAFKA_CLIENT_BASE64_DECODING_ENABLED, false);
+    kafkaClientBase64DecodingGuardEnabled =
+        configProvider.getBoolean(KAFKA_CLIENT_BASE64_DECODING_GUARD_ENABLED, true);
     jmsPropagationEnabled = isPropagationEnabled(true, "jms");
     jmsPropagationDisabledTopics =
         tryMakeImmutableSet(configProvider.getList(JMS_PROPAGATION_DISABLED_TOPICS));
@@ -5103,6 +5107,10 @@ public boolean isKafkaClientBase64DecodingEnabled() {
     return kafkaClientBase64DecodingEnabled;
   }
 
+  public boolean isKafkaClientBase64DecodingGuardEnabled() {
+    return kafkaClientBase64DecodingGuardEnabled;
+  }
+
   public boolean isRabbitPropagationEnabled() {
     return rabbitPropagationEnabled;
   }
@@ -6852,6 +6860,8 @@ public String toString() {
         + kafkaClientPropagationDisabledTopics
         + ", kafkaClientBase64DecodingEnabled="
         + kafkaClientBase64DecodingEnabled
+        + ", kafkaClientBase64DecodingGuardEnabled="
+        + kafkaClientBase64DecodingGuardEnabled
         + ", jmsPropagationEnabled="
         + jmsPropagationEnabled
         + ", jmsPropagationDisabledTopics="

@@ -188,4 +188,95 @@ public T apply(Object input) {
           return null;
         }
       };
+
+  /**
+   * Base64-decodes bytes that occasionally aren't valid Base64 at all (e.g. a misconfigured
+   * upstream producer sending mixed encodings), without paying for {@link Base64}'s decoder
+   * throwing and filling in a stack trace on every one of those failures.
+   *
+   * <p>Tracks a simple hysteresis "breaker": after a real decode failure it switches into a guarded
+   * state and cheaply pre-checks the Base64 alphabet before calling into {@link Base64}'s decoder,
+   * until enough consecutive inputs look valid again to close the guard. A precheck-detected
+   * failure throws {@link DefinitelyNotBase64Exception}, a stack-trace-free stand-in, so the {@link
+   * #decode} contract still matches {@link Base64}'s own (throws {@link IllegalArgumentException}
+   * on failure); a failure the precheck misses still throws the JDK's own exception, unmodified,
+   * with its real message and stack trace.
+   *
+   * <p>Not safe for concurrent use by multiple threads expecting independent guard state — the
+   * guard state is a plain (non-volatile, non-atomic) field on purpose: it's advisory hysteresis,
+   * not correctness-critical, so a stale read across threads just costs one extra precheck or one
+   * extra real exception.
+   */
+  public static final class GuardedBase64Decode {
+
+    private static final int CLOSE_THRESHOLD = 20;
+    private static final boolean[] BASE64_ALPHABET = buildAlphabetTable();
+
+    private static boolean[] buildAlphabetTable() {
+      boolean[] table = new boolean[256];
+      String alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=";
+      for (int i = 0; i < alphabet.length(); i++) {
+        table[alphabet.charAt(i)] = true;
+      }
+      return table;
+    }
+
+    // Branch-free per byte on purpose: a data-dependent early exit helps the rare "bad byte
+    // early" case but hurts the common valid case, which always scans the whole buffer anyway.
+    private static boolean looksLikeBase64(byte[] bytes) {
+      if (bytes.length == 0 || (bytes.length & 3) != 0) {
+        return false;
+      }
+      boolean valid = true;
+      for (byte b : bytes) {
+        valid &= BASE64_ALPHABET[b & 0xFF];
+      }
+      return valid;
+    }
+
+    /** Thrown only when the alphabet precheck already knows decoding would fail. */
+    public static final class DefinitelyNotBase64Exception extends IllegalArgumentException {
+      private static final String MESSAGE = "Input is not valid Base64";
+
+      DefinitelyNotBase64Exception() {
+        super(MESSAGE);
+      }
+
+      // No caller needs a stack trace for a failure we already knew about before calling into
+      // Base64's decoder, so skip the walk that gives java.util.Base64's own exception its cost.
+      @Override
+      public synchronized Throwable fillInStackTrace() {
+        return this;
+      }
+    }
+
+    // Single-word countdown: 0 == closed (no precheck), >0 == guarded, counting down to close.
+    private int state;
+
+    /** Decodes {@code bytes}, throwing {@link IllegalArgumentException} on failure. */
+    public String decode(byte[] bytes) {
+      if (state > 0 && !looksLikeBase64(bytes)) {
+        throw new DefinitelyNotBase64Exception();
+      }
+      try {
+        String result = new String(Base64.getDecoder().decode(bytes), UTF_8);
+        if (state > 0) {
+          state--;
+        }
+        return result;
+      } catch (IllegalArgumentException e) {
+        state = CLOSE_THRESHOLD;
+        throw e;
+      }
+    }
+
+    /** Decodes {@code bytes}, returning {@code null} on failure instead of throwing. */
+    public String decodeOrNull(byte[] bytes) {
+      try {
+        return decode(bytes);
+      } catch (IllegalArgumentException e) {
+        return null;
+      }
+    }
+  }
 }
@@ -0,0 +1,86 @@
+package datadog.trace.api;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import datadog.trace.api.Functions.GuardedBase64Decode;
+import datadog.trace.api.Functions.GuardedBase64Decode.DefinitelyNotBase64Exception;
+import java.util.Base64;
+import org.junit.jupiter.api.Test;
+
+class GuardedBase64DecodeTest {
+
+  private static final byte[] VALID =
+      Base64.getEncoder().encode("x-datadog-trace-id".getBytes(UTF_8));
+  private static final byte[] INVALID = "not-valid-base64!@#".getBytes(UTF_8);
+
+  @Test
+  void decodeReturnsDecodedStringForValidInput() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertEquals("x-datadog-trace-id", decode.decode(VALID));
+  }
+
+  @Test
+  void decodeThrowsRealExceptionForInvalidInputWhenNotGuarded() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> decode.decode(INVALID));
+    // the guard hasn't seen a failure yet, so this must be the JDK's own exception, not our
+    // stack-trace-free stand-in
+    assertEquals(IllegalArgumentException.class, e.getClass());
+  }
+
+  @Test
+  void decodeThrowsFastFailStandInWhileGuarded() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertThrows(IllegalArgumentException.class, () -> decode.decode(INVALID));
+
+    DefinitelyNotBase64Exception e =
+        assertThrows(DefinitelyNotBase64Exception.class, () -> decode.decode(INVALID));
+    assertEquals(0, e.getStackTrace().length);
+  }
+
+  @Test
+  void decodeStillDecodesValidInputWhileGuarded() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertThrows(IllegalArgumentException.class, () -> decode.decode(INVALID));
+
+    assertEquals("x-datadog-trace-id", decode.decode(VALID));
+  }
+
+  @Test
+  void decodeClosesGuardAfterEnoughValidInput() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertThrows(IllegalArgumentException.class, () -> decode.decode(INVALID));
+
+    for (int i = 0; i < 50; i++) {
+      decode.decode(VALID);
+    }
+
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> decode.decode(INVALID));
+    // the guard has closed again, so this should be a real decode failure, not the stand-in
+    assertEquals(IllegalArgumentException.class, e.getClass());
+  }
+
+  @Test
+  void decodeOrNullReturnsDecodedStringForValidInput() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertEquals("x-datadog-trace-id", decode.decodeOrNull(VALID));
+  }
+
+  @Test
+  void decodeOrNullReturnsNullForInvalidInput() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertNull(decode.decodeOrNull(INVALID));
+  }
+
+  @Test
+  void decodeOrNullReturnsNullWhileGuarded() {
+    GuardedBase64Decode decode = new GuardedBase64Decode();
+    assertNull(decode.decodeOrNull(INVALID));
+    assertNull(decode.decodeOrNull(INVALID));
+  }
+}
@@ -2300,6 +2300,14 @@
         "aliases": []
       }
     ],
+    "DD_KAFKA_CLIENT_BASE64_DECODING_GUARD_ENABLED": [
+      {
+        "version": "A",
+        "type": "boolean",
+        "default": "true",
+        "aliases": []
+      }
+    ],
     "DD_KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS": [
       {
         "version": "A",