This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit c2681b229b32863d0e067a4b55421b25c7b995f8 Author: Vova Kolmakov <[email protected]> AuthorDate: Tue Jun 16 15:20:11 2026 +0700 [MINOR] Wait for ZK connection in lock provider to de-flake direct-marker detection test (#19014) (cherry picked from commit f3ccf1b3c31b52e7ece2d596a7749369e2f7e0b0) --- .../lock/BaseZookeeperBasedLockProvider.java | 27 ++++++++++++++++++-- .../TestZookeeperBasedLockProvider.java | 21 ++++++++++++++++ ...erBasedDetectionStrategyWithZKLockProvider.java | 29 +++++++++++++++++++--- 3 files changed, 71 insertions(+), 6 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java index d5b04c15c005..6cdee60e2d1b 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java @@ -68,6 +68,7 @@ public abstract class BaseZookeeperBasedLockProvider implements LockProvider<Int this.lockConfiguration = lockConfiguration; zkBasePath = getZkBasePath(lockConfiguration); lockKey = getLockKey(lockConfiguration); + int connectionTimeoutMs = ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECTION_TIMEOUT_MS); this.curatorFrameworkClient = CuratorFrameworkFactory.builder() .connectString(ConfigUtils.getStringWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECT_URL)) .retryPolicy(new BoundedExponentialBackoffRetry( @@ -75,10 +76,32 @@ public abstract class BaseZookeeperBasedLockProvider implements LockProvider<Int ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS), ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), LOCK_ACQUIRE_NUM_RETRIES))) .sessionTimeoutMs(ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_SESSION_TIMEOUT_MS)) - .connectionTimeoutMs(ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECTION_TIMEOUT_MS)) + .connectionTimeoutMs(connectionTimeoutMs) .build(); this.curatorFrameworkClient.start(); - createPathIfNotExists(); + // Once started, the Curator client owns background threads. If anything below throws, the + // constructor never returns the instance, so the caller can never invoke close() - clean up here. + try { + if (!this.curatorFrameworkClient.blockUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS)) { + throw new HoodieLockException("Failed to connect to ZooKeeper within " + connectionTimeoutMs + " ms"); + } + createPathIfNotExists(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + closeQuietly(); + throw new HoodieLockException("Interrupted while waiting to connect to ZooKeeper", e); + } catch (RuntimeException e) { + closeQuietly(); + throw e; + } + } + + private void closeQuietly() { + try { + this.curatorFrameworkClient.close(); + } catch (Exception ex) { + log.warn("Failed to close ZooKeeper client after failed initialization", ex); + } } protected abstract String getZkBasePath(LockConfiguration lockConfiguration); diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestZookeeperBasedLockProvider.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestZookeeperBasedLockProvider.java index fab3dee8f8f6..fa1a8329b65f 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestZookeeperBasedLockProvider.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestZookeeperBasedLockProvider.java @@ -42,6 +42,7 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import java.io.IOException; +import java.time.Duration; import java.util.Properties; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; @@ -194,4 +195,24 @@ public class TestZookeeperBasedLockProvider { ZookeeperBasedLockProvider zookeeperBasedLockProvider = new ZookeeperBasedLockProvider(zkConfWithZkBasePathAndLockKeyLock, null); zookeeperBasedLockProvider.unlock(); } + + @Test + public void testFailFastWhenZkUnreachable() { + Properties properties = new Properties(); + // Nothing listens on 127.0.0.1:1, so the connect-wait must time out instead of hanging. + properties.setProperty(ZK_CONNECT_URL_PROP_KEY, "127.0.0.1:1"); + properties.setProperty(ZK_BASE_PATH_PROP_KEY, basePath); + properties.setProperty(ZK_LOCK_KEY_PROP_KEY, key); + properties.setProperty(LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "100"); + properties.setProperty(LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS_PROP_KEY, "300"); + properties.setProperty(LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "1"); + properties.setProperty(ZK_SESSION_TIMEOUT_MS_PROP_KEY, "1000"); + properties.setProperty(ZK_CONNECTION_TIMEOUT_MS_PROP_KEY, "1000"); + LockConfiguration unreachable = new LockConfiguration(properties); + // Construction must fail fast (seconds, bounded by the connection timeout) with a + // HoodieLockException instead of being amplified into a multi-minute retry hang. + Assertions.assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + Assertions.assertThrows(HoodieLockException.class, + () -> new ZookeeperBasedLockProvider(unreachable, null))); + } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockProvider.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockProvider.java index deb551e9d245..a2a5a92f69eb 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockProvider.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockProvider.java @@ -51,9 +51,16 @@ import java.io.IOException; import java.util.List; import java.util.Properties; +import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY; import static org.apache.hudi.common.config.LockConfiguration.ZK_BASE_PATH_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.ZK_CONNECTION_TIMEOUT_MS_PROP_KEY; import static org.apache.hudi.common.config.LockConfiguration.ZK_CONNECT_URL_PROP_KEY; import static org.apache.hudi.common.config.LockConfiguration.ZK_LOCK_KEY_PROP_KEY; +import static org.apache.hudi.common.config.LockConfiguration.ZK_SESSION_TIMEOUT_MS_PROP_KEY; import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -82,6 +89,15 @@ public class TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockPr properties.setProperty(ZK_CONNECT_URL_PROP_KEY, server.getConnectString()); properties.setProperty(ZK_BASE_PATH_PROP_KEY, server.getTempDirectory().getAbsolutePath()); properties.setProperty(ZK_LOCK_KEY_PROP_KEY, "key"); + // Bound lock retries and ZK timeouts so a transient connection failure fails fast in seconds + // instead of being amplified by the production-default retry layers into a multi-minute hang. + properties.setProperty(LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000"); + properties.setProperty(LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS_PROP_KEY, "3000"); + properties.setProperty(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY, "3"); + properties.setProperty(LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3"); + properties.setProperty(ZK_SESSION_TIMEOUT_MS_PROP_KEY, "10000"); + properties.setProperty(ZK_CONNECTION_TIMEOUT_MS_PROP_KEY, "10000"); + properties.setProperty(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "1000"); config = getConfigBuilder() .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder() @@ -104,10 +120,15 @@ public class TestSimpleTransactionDirectMarkerBasedDetectionStrategyWithZKLockPr @AfterEach public void clean() throws IOException { - cleanupResources(); - FileIOUtils.deleteDirectory(new File(basePath)); - if (server != null) { - server.close(); + try { + cleanupResources(); + FileIOUtils.deleteDirectory(new File(basePath)); + } finally { + // Always stop the embedded ZooKeeper server, even if resource cleanup or directory + // deletion above throws, so the server is not leaked across parameterized runs. + if (server != null) { + server.close(); + } } }
