Fix MDC connector context not propagating to async querier thread

Since v10.8.5, logs from the async query path (TableQuerierProcessor,
TableQuerier, etc.) lost the [connector-name|task-N] prefix because
Kafka Connect's MDC is thread-local and was not propagated to the
background executor thread introduced by the RecordQueue architecture.

Capture MDC.getCopyOfContextMap() on the calling (poll) thread before
submitting to the executor, then restore it on the executor thread
before ConnectLogContext is created. This allows ConnectLogContext to
find and extend the connector.context MDC entry as intended. MDC is
cleared in a finally block to prevent leaking to pooled threads.

Author: sangeet259

src/main/java/io/confluent/connect/jdbc/util/RecordQueue.java

@@ -13,6 +13,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
@@ -29,6 +30,8 @@
 import java.util.function.IntFunction;
 import java.util.function.Supplier;
 
+import org.slf4j.MDC;
+
 /**
  * A concurrent and blocking queue for source records, with the ability to asynchronously execute
  * record generators that send records to this queue.
@@ -318,12 +321,17 @@ public <T> CompletableFuture<T> submit(
     final RecordDestination<RecordT> cancellableDestination =
         this.withAdditionalRunningCondition(stillRun::get);
 
+    // Capture the caller's MDC (which includes Connect's connector.context)
+    // so it can be restored on the executor thread
+    Map<String, String> callerMdc = MDC.getCopyOfContextMap();
+
     // Submit the function and use the cancellable destination
     Supplier<T> supplier = createLoggingSupplier(
         operationName,
         logContext,
         cancellableDestination,
-        generatorProcessor
+        generatorProcessor,
+        callerMdc
     );
     return CompletableFuture.supplyAsync(supplier, executor);
   }
@@ -465,9 +473,15 @@ protected <T> Supplier<T> createLoggingSupplier(
       String operationName,
       String logContext,
       RecordDestination<RecordT> destination,
-      Function<RecordDestination<RecordT>, T> generatorProcessor
+      Function<RecordDestination<RecordT>, T> generatorProcessor,
+      Map<String, String> callerMdc
   ) {
     return () -> {
+      // Restore the caller's MDC on this executor thread so that
+      // ConnectLogContext can find and extend the connector.context
+      if (callerMdc != null) {
+        MDC.setContextMap(callerMdc);
+      }
       try (ConnectLogContext context = new ConnectLogContext(logContext)) {
         try {
           log.debug("{}Starting {}", context.prefix(), operationName);
@@ -479,6 +493,9 @@ protected <T> Supplier<T> createLoggingSupplier(
         } finally {
           log.debug("{}Stopped {}", context.prefix(), operationName);
         }
+      } finally {
+        // Clean up MDC to prevent leaking context to pooled threads
+        MDC.clear();
       }
     };
   }

src/test/java/io/confluent/connect/jdbc/util/ConnectLogContextTest.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright [2024 - 2024] Confluent Inc.
+ */
+
+package io.confluent.connect.jdbc.util;
+
+import org.junit.After;
+import org.junit.Test;
+import org.slf4j.MDC;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class ConnectLogContextTest {
+
+  private static final String CONNECTOR_MDC_CONTEXT = "connector.context";
+
+  @After
+  public void tearDown() {
+    MDC.clear();
+  }
+
+  @Test
+  public void shouldAppendSuffixWhenMdcHasConnectorContext() {
+    MDC.put(CONNECTOR_MDC_CONTEXT, "[my-connector|task-0] ");
+
+    try (ConnectLogContext ctx = new ConnectLogContext("tableQuerierProcessor")) {
+      // MDC should now have the appended context
+      assertEquals(
+          "[my-connector|task-0|tableQuerierProcessor] ",
+          MDC.get(CONNECTOR_MDC_CONTEXT)
+      );
+      // prefix should be empty since MDC is used
+      assertEquals("", ctx.prefix());
+    }
+
+    // After close, MDC should be restored to original value
+    assertEquals("[my-connector|task-0] ", MDC.get(CONNECTOR_MDC_CONTEXT));
+  }
+
+  @Test
+  public void shouldUsePrefixWhenMdcIsEmpty() {
+    // No MDC set
+    try (ConnectLogContext ctx = new ConnectLogContext("tableQuerierProcessor")) {
+      // MDC should still be null
+      assertNull(MDC.get(CONNECTOR_MDC_CONTEXT));
+      // prefix should contain the log context
+      assertEquals("tableQuerierProcessor ", ctx.prefix());
+    }
+  }
+
+  @Test
+  public void shouldHandleNestedContexts() {
+    MDC.put(CONNECTOR_MDC_CONTEXT, "[my-connector|task-0] ");
+
+    try (ConnectLogContext outer = new ConnectLogContext("outer")) {
+      assertEquals("[my-connector|task-0|outer] ", MDC.get(CONNECTOR_MDC_CONTEXT));
+
+      try (ConnectLogContext inner = new ConnectLogContext("inner")) {
+        assertEquals("[my-connector|task-0|outer|inner] ", MDC.get(CONNECTOR_MDC_CONTEXT));
+      }
+
+      // After inner closes, should restore to outer's context
+      assertEquals("[my-connector|task-0|outer] ", MDC.get(CONNECTOR_MDC_CONTEXT));
+    }
+
+    // After outer closes, should restore to original
+    assertEquals("[my-connector|task-0] ", MDC.get(CONNECTOR_MDC_CONTEXT));
+  }
+}

src/test/java/io/confluent/connect/jdbc/util/RecordQueueMdcTest.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright [2024 - 2024] Confluent Inc.
+ */
+
+package io.confluent.connect.jdbc.util;
+
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.MDC;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class RecordQueueMdcTest {
+
+  private static final String CONNECTOR_MDC_CONTEXT = "connector.context";
+
+  private RecordQueue<SourceRecord> queue;
+
+  @Before
+  public void setUp() {
+    queue = RecordQueue.<SourceRecord>builder()
+        .maxBatchSize(10)
+        .maxExecutorThreads(1)
+        .build();
+  }
+
+  @After
+  public void tearDown() throws InterruptedException {
+    MDC.clear();
+    if (queue != null) {
+      queue.stop();
+      queue.awaitTermination(Duration.ofSeconds(5));
+    }
+  }
+
+  @Test
+  public void shouldPropagateConnectorContextToExecutorThread() throws Exception {
+    // Simulate Kafka Connect setting MDC on the calling thread
+    MDC.put(CONNECTOR_MDC_CONTEXT, "[my-connector|task-0] ");
+
+    AtomicReference<String> capturedMdc = new AtomicReference<>();
+
+    CompletableFuture<Boolean> future = queue.submit(
+        "Test Operation",
+        "testProcessor",
+        destination -> {
+          // Capture the MDC value on the executor thread
+          capturedMdc.set(MDC.get(CONNECTOR_MDC_CONTEXT));
+          return true;
+        }
+    );
+
+    assertTrue(future.get(10, TimeUnit.SECONDS));
+
+    // The executor thread should have seen the connector context with appended suffix
+    assertNotNull("MDC should be propagated to executor thread", capturedMdc.get());
+    assertEquals(
+        "[my-connector|task-0|testProcessor] ",
+        capturedMdc.get()
+    );
+  }
+
+  @Test
+  public void shouldCleanUpMdcAfterExecution() throws Exception {
+    MDC.put(CONNECTOR_MDC_CONTEXT, "[my-connector|task-0] ");
+
+    AtomicReference<String> mdcAfterExecution = new AtomicReference<>();
+
+    // First submit — sets MDC on executor thread
+    CompletableFuture<Boolean> future1 = queue.submit(
+        "First Operation",
+        "firstProcessor",
+        destination -> true
+    );
+    future1.get(10, TimeUnit.SECONDS);
+
+    // Clear caller's MDC to simulate a different calling context
+    MDC.clear();
+
+    // Second submit — without MDC on the calling thread
+    CompletableFuture<Boolean> future2 = queue.submit(
+        "Second Operation",
+        "secondProcessor",
+        destination -> {
+          mdcAfterExecution.set(MDC.get(CONNECTOR_MDC_CONTEXT));
+          return true;
+        }
+    );
+    future2.get(10, TimeUnit.SECONDS);
+
+    // The second execution should NOT have the first execution's MDC
+    // (it should be null since the caller had no MDC set)
+    assertNull(
+        "MDC should be cleaned up between executions on pooled threads",
+        mdcAfterExecution.get()
+    );
+  }
+
+  @Test
+  public void shouldWorkWhenCallerHasNoMdc() throws Exception {
+    // No MDC set on the calling thread
+    AtomicReference<String> capturedMdc = new AtomicReference<>();
+
+    CompletableFuture<Boolean> future = queue.submit(
+        "Test Operation",
+        "testProcessor",
+        destination -> {
+          capturedMdc.set(MDC.get(CONNECTOR_MDC_CONTEXT));
+          return true;
+        }
+    );
+
+    assertTrue(future.get(10, TimeUnit.SECONDS));
+
+    // Without MDC, ConnectLogContext falls back to prefix mode — no MDC set
+    assertNull(capturedMdc.get());
+  }
+}