This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12480-8b28ff8e7c6a5105e033909ba53b152df0289255 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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); + } }
