This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 53733a112c [Fix][Connector-V2] Fix Kudu client deadlock in Flink
JobManager on JDK 11 (#12480)
53733a112c is described below
commit 53733a112c9e935c5d97f7f24be249caa43b4fd5
Author: Goutam Adwant <[email protected]>
AuthorDate: Wed Sep 30 13:18:47 2026 +0000
[Fix][Connector-V2] Fix Kudu client deadlock in Flink JobManager on JDK 11
(#12480)
---
.../connectors/seatunnel/kudu/util/KuduUtil.java | 28 ++++++-
.../kudu/kuduclient/KuduClientResourceTest.java | 94 ++++++++++++++++++++++
2 files changed, 121 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
index 3faded4a5e..4b1f4239d5 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
@@ -43,7 +43,12 @@ import lombok.extern.slf4j.Slf4j;
import sun.security.krb5.Config;
import sun.security.krb5.KrbException;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+
import java.io.IOException;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.List;
@@ -161,7 +166,28 @@ public class KuduUtil {
if (executorService != null) {
builder.nioExecutor(executorService);
}
- return builder.build().syncClient();
+ return buildOutsideUnusedCallerSubject(builder).syncClient();
+ }
+
+ /**
+ * Builds the client outside the caller's Subject when that Subject has no
Kerberos principal.
+ *
+ * <p>Kudu ignores such a Subject, but its SecurityContext still calls
{@link
+ * Subject#toString()} on it. That method holds the Subject's principal
set lock while it
+ * formats the principals, and on JDK 11 the first UnixPrincipal#toString
loads a resource
+ * bundle through SubjectDomainCombiner#combine, which needs the combiner
lock. Thread creation
+ * under the same Subject takes these two locks in the opposite order, so
the call can deadlock.
+ * Flink runs its JobManager and TaskManager inside such a Subject, and
the deadlock froze the
+ * whole JobManager. Kudu makes the same choice without a caller Subject,
so behavior is
+ * unchanged.
+ */
+ private static AsyncKuduClient buildOutsideUnusedCallerSubject(
+ AsyncKuduClient.AsyncKuduClientBuilder builder) {
+ Subject subject = Subject.getSubject(AccessController.getContext());
+ if (subject == null ||
!subject.getPrincipals(KerberosPrincipal.class).isEmpty()) {
+ return builder.build();
+ }
+ return Subject.doAs(null, (PrivilegedAction<AsyncKuduClient>)
builder::build);
}
public static List<KuduScanToken> getKuduScanToken(
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
index 4b2cb2ad4e..ba10d8a94e 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
@@ -17,6 +17,10 @@
package org.apache.seatunnel.connectors.seatunnel.kudu.kuduclient;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.kudu.config.CommonConfig;
+import org.apache.seatunnel.connectors.seatunnel.kudu.util.KuduUtil;
+
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
@@ -25,8 +29,17 @@ import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
import org.mockito.Mockito;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.security.Principal;
+import java.security.PrivilegedExceptionAction;
+import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
class KuduClientResourceTest {
@@ -79,4 +92,85 @@ class KuduClientResourceTest {
Mockito.verify(executorService).shutdown();
Mockito.verify(executorService).awaitTermination(20L,
TimeUnit.SECONDS);
}
+
+ /**
+ * Flink runs its JobManager and TaskManager inside a Subject without
Kerberos credentials. Kudu
+ * ignores such a Subject, but it still calls Subject#toString on it while
holding the Subject's
+ * principal set lock. On JDK 11 that call can deadlock with any thread
being created under the
+ * same Subject, which froze the Flink JobManager in the Kudu E2E tests.
+ */
+ @Test
+ void shouldNotInspectCallerSubjectWithoutKerberosCredentials() throws
Exception {
+ AtomicInteger toStringCalls = new AtomicInteger();
+ Principal principal =
+ new Principal() {
+ @Override
+ public String getName() {
+ return "flink";
+ }
+
+ @Override
+ public String toString() {
+ toStringCalls.incrementAndGet();
+ return getName();
+ }
+ };
+ Subject subject =
+ new Subject(
+ false,
+ Collections.singleton(principal),
+ Collections.emptySet(),
+ Collections.emptySet());
+ CommonConfig config =
+ new CommonConfig(
+ ReadonlyConfig.fromMap(
+ Collections.singletonMap("kudu_masters",
"localhost:7051")));
+
+ KuduClientResource resource =
+ Subject.doAs(
+ subject,
+ (PrivilegedExceptionAction<KuduClientResource>)
+ () -> KuduUtil.getKuduClientResource(config));
+ resource.close();
+
+ Assertions.assertEquals(0, toStringCalls.get());
+ }
+
+ @Test
+ void shouldPassCallerSubjectWithKerberosPrincipalToKudu() throws Exception
{
+ Subject subject =
+ new Subject(
+ false,
+ Collections.singleton(new
KerberosPrincipal("[email protected]")),
+ Collections.emptySet(),
+ Collections.emptySet());
+
+ KuduClientResource resource =
+ Subject.doAs(
+ subject,
+ (PrivilegedExceptionAction<KuduClientResource>)
+ () ->
KuduUtil.getKuduClientResource(localConfig()));
+ try {
+ Assertions.assertSame(subject, kuduSubject(resource.getClient()));
+ } finally {
+ resource.close();
+ }
+ }
+
+ private static CommonConfig localConfig() {
+ return new CommonConfig(
+
ReadonlyConfig.fromMap(Collections.singletonMap("kudu_masters",
"localhost:7051")));
+ }
+
+ private static Subject kuduSubject(KuduClient client) throws Exception {
+ Field asyncClientField =
KuduClient.class.getDeclaredField("asyncClient");
+ asyncClientField.setAccessible(true);
+ Object asyncClient = asyncClientField.get(client);
+ Field securityContextField =
asyncClient.getClass().getDeclaredField("securityContext");
+ securityContextField.setAccessible(true);
+ Object securityContext = securityContextField.get(asyncClient);
+ Method getSubject =
securityContext.getClass().getDeclaredMethod("getSubject");
+ getSubject.setAccessible(true);
+ return (Subject) getSubject.invoke(securityContext);
+ }
}
