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 af0fbfea603a8eecc6b72388c8d5fb160f175d42 Author: voonhous <[email protected]> AuthorDate: Thu Jun 25 19:50:36 2026 +0800 test(spark): de-flake TestHoodieClientMultiWriter early-conflict detection (#19069) The early-conflict-detection test's commit-004 path waited heartBeatIntervalForCommit4 * 2 (6s) for the failed commit 003's heartbeat to expire, which assumed the default hoodie.client.heartbeat.tolerable.misses of 2 (expiry window = misses * interval = 2 * 3000 = 6000ms). #18904 raised that default to 10, growing the window to 30000ms, so commit 004 saw 003 as still alive and hit a false early conflict, failing assertDoesNotThrow. Because that failure threw before the un-finally'd TestingServer was closed, and surefire runs the shard with reuseForks=true, the leaked embedded ZooKeeper kept holding its admin server's fixed port 8080, so every later rerun failed with "Unable to instantiate ZookeeperBasedLockProvider" (BindException: Address already in use). Fixes: - Pin heartbeat tolerable misses (constant EARLY_CONFLICT_HEARTBEAT_TOLERABLE_MISSES) in buildWriteConfigForEarlyConflictDetect so the wait no longer depends on the global default, and derive the wait as interval * (misses + 1) for a clear margin. - Disable ZooKeeper's unused embedded admin server (zookeeper.admin.enableServer=false) to remove the fixed-port-8080 collision. - Move the TestingServer into a field closed in @AfterEach so a failing assertion cannot leak it across the reused fork. Closes #19068 (cherry picked from commit 08ba3f7e4e80fa8c55d8fd79e04c9944c6df2d27) --- .../hudi/client/TestHoodieClientMultiWriter.java | 37 ++++++++++++++++------ 1 file changed, 28 insertions(+), 9 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java index f07414f18af1..bebdf9d88e60 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java +++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java @@ -137,8 +137,21 @@ import static org.junit.jupiter.api.Assertions.fail; @Tag("functional") public class TestHoodieClientMultiWriter extends HoodieClientTestBase { + // Pin the tolerable heartbeat misses used by the early-conflict-detection test so its + // heartbeat-expiry wait does not depend on the global default + // (hoodie.client.heartbeat.tolerable.misses), which changed from 2 to 10 in #18904. + private static final int EARLY_CONFLICT_HEARTBEAT_TOLERABLE_MISSES = 2; + + static { + // ZooKeeper's embedded admin server binds the fixed default port 8080 (Curator only + // randomizes the client port), so a concurrent or leaked TestingServer in a reused + // fork collides with "Address already in use". The admin server is unused here. + System.setProperty("zookeeper.admin.enableServer", "false"); + } + private Properties lockProperties = null; + private TestingServer zkTestingServer = null; /** * super is not thread safe!! @@ -175,6 +188,10 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { @AfterEach public void clean() throws IOException { + if (zkTestingServer != null) { + zkTestingServer.close(); + zkTestingServer = null; + } cleanupResources(); } @@ -462,14 +479,14 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { int heartBeatIntervalForCommit4 = 3 * 1000; HoodieWriteConfig writeConfig; - TestingServer server = null; if (earlyConflictDetectionStrategy.equalsIgnoreCase(SimpleTransactionDirectMarkerBasedDetectionStrategy.class.getName())) { // need to setup zk related env there. Bcz SimpleTransactionDirectMarkerBasedDetectionStrategy is only support zk lock for now. - server = new TestingServer(); + // zkTestingServer is closed in @AfterEach so a failing assertion cannot leak it (and its port-8080 admin server). + zkTestingServer = new TestingServer(); Properties properties = new Properties(); properties.setProperty(ZK_BASE_PATH_PROP_KEY, basePath); - properties.setProperty(ZK_CONNECT_URL_PROP_KEY, server.getConnectString()); - properties.setProperty(ZK_BASE_PATH_PROP_KEY, server.getTempDirectory().getAbsolutePath()); + properties.setProperty(ZK_CONNECT_URL_PROP_KEY, zkTestingServer.getConnectString()); + properties.setProperty(ZK_BASE_PATH_PROP_KEY, zkTestingServer.getTempDirectory().getAbsolutePath()); properties.setProperty(ZK_SESSION_TIMEOUT_MS_PROP_KEY, "10000"); properties.setProperty(ZK_CONNECTION_TIMEOUT_MS_PROP_KEY, "10000"); properties.setProperty(ZK_LOCK_KEY_PROP_KEY, "key"); @@ -521,8 +538,11 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { storage.create(heartbeatFilePath, true); // Wait for heart beat expired for failed commitTime3 "003" - // Otherwise commit4 still can see conflict between failed write 003. - Thread.sleep(heartBeatIntervalForCommit4 * 2); + // Otherwise commit4 still can see conflict between failed write 003. The early-conflict + // check treats 003 as alive until its heartbeat is older than + // (tolerable misses * heartbeat interval); tolerable misses is pinned in + // buildWriteConfigForEarlyConflictDetect, so wait one interval past that window. + Thread.sleep(heartBeatIntervalForCommit4 * (EARLY_CONFLICT_HEARTBEAT_TOLERABLE_MISSES + 1)); final String nextCommitTime4 = "004"; assertDoesNotThrow(() -> { @@ -541,9 +561,6 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { assertTrue(completedInstant.contains(nextCommitTime4)); FileIOUtils.deleteDirectory(new File(basePath)); - if (server != null) { - server.close(); - } client1.close(); client2.close(); client3.close(); @@ -1771,6 +1788,7 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { if (markerType.equalsIgnoreCase(MarkerType.DIRECT.name())) { return getConfigBuilder() .withHeartbeatIntervalInMs(60 * 1000) + .withHeartbeatTolerableMisses(EARLY_CONFLICT_HEARTBEAT_TOLERABLE_MISSES) .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder() .withStorageType(FileSystemViewStorageType.MEMORY) .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build()) @@ -1791,6 +1809,7 @@ public class TestHoodieClientMultiWriter extends HoodieClientTestBase { return getConfigBuilder() .withStorageConfig(HoodieStorageConfig.newBuilder().parquetMaxFileSize(20 * 1024).build()) .withHeartbeatIntervalInMs(60 * 1000) + .withHeartbeatTolerableMisses(EARLY_CONFLICT_HEARTBEAT_TOLERABLE_MISSES) .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder() .withStorageType(FileSystemViewStorageType.MEMORY) .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build())
