This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch release-2.4 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7818f82ad8371cb47c3742c9a11b9d1b90015206 Author: Dale Lane <[email protected]> AuthorDate: Tue Oct 20 16:02:12 2026 +0100 [FLINK-41048][container] Fix MiniClusterJobEntryPointITCase race condition To be able to submit queries to the MiniCluster, I'd written the test to pick an available port number, then give that to the minicluster to bind to. Then I could use that port number when submitting queries. The problem is that it introduced a race condition. In the window between identifying an available port, and starting a MiniCluster to bind to it, something else could grab the port. I can't recreate this locally. Presumably something about the CI environment being more constrained widened the window, or something about the Centos 7 (3.10) version the CI workers use changes how port numbers are dished out. Either way, it is not safe to assume that a port number that's available when the test starts will remain so for the duration of the test. I've changed the minicluster config to use bind port 0 so it chooses a free port by itself. This does add some complexity to the test, because it means I have to go and discover what port it used after it's started to be able to submit queries. Signed-off-by: Dale Lane <[email protected]> (cherry picked from commit e56648e7ea288bb0d7b5f9e4e1ee18fc9bb4f068) --- .../entrypoint/MiniClusterJobEntryPointITCase.java | 65 +++++++++++++++------- 1 file changed, 46 insertions(+), 19 deletions(-) diff --git a/flink-container/src/test/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPointITCase.java b/flink-container/src/test/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPointITCase.java index 0714bbe2b32..7c2f98afa2f 100644 --- a/flink-container/src/test/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPointITCase.java +++ b/flink-container/src/test/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPointITCase.java @@ -59,6 +59,7 @@ import org.apache.flink.runtime.highavailability.HighAvailabilityServices; import org.apache.flink.runtime.highavailability.HighAvailabilityServicesFactory; import org.apache.flink.runtime.highavailability.JobResultStoreOptions; import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServices; +import org.apache.flink.runtime.leaderelection.LeaderInformation; import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup; import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; import org.apache.flink.runtime.state.LocalRecoveryConfig; @@ -68,14 +69,17 @@ import org.apache.flink.runtime.state.changelog.inmemory.InMemoryStateChangelogS import org.apache.flink.runtime.taskexecutor.TaskManagerRunner; import org.apache.flink.runtime.util.Hardware; import org.apache.flink.runtime.util.TestingFatalErrorHandler; +import org.apache.flink.runtime.util.ZooKeeperUtils; import org.apache.flink.runtime.zookeeper.ZooKeeperExtension; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; import org.apache.flink.util.NetUtils; import org.apache.flink.util.jackson.JacksonMapperFactory; +import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -152,7 +156,6 @@ class MiniClusterJobEntryPointITCase { private Path markerFile; private JobID jobId; private Path log4jConfig; - private int restPort; private String haClusterId; @BeforeEach @@ -160,9 +163,6 @@ class MiniClusterJobEntryPointITCase { confDir = Files.createDirectories(tempDir.resolve("conf")); markerFile = tempDir.resolve("main-invocations"); jobId = JobID.generate(); - try (NetUtils.Port port = NetUtils.getAvailablePort()) { - restPort = port.getPort(); - } log4jConfig = tempDir.resolve("log4j2.properties"); Files.write( @@ -192,8 +192,7 @@ class MiniClusterJobEntryPointITCase { entry(JobResultStoreOptions.DELETE_ON_COMMIT, false), entry(ApplicationResultStoreOptions.DELETE_ON_COMMIT, false), entry(RestOptions.ADDRESS, "localhost"), - entry(RestOptions.PORT, restPort), - entry(RestOptions.BIND_PORT, restPort))); + entry(RestOptions.BIND_PORT, 0))); } @Test @@ -479,14 +478,14 @@ class MiniClusterJobEntryPointITCase { // a separate HA cluster, as for a job moved to a new deployment, so that the job can only // be restored from the savepoint and not recovered from HA metadata + haClusterId = haClusterId + "-restored"; final List<String> arguments = new ArrayList<>( List.of( "--fromSavepoint", savepointPath, dynamicProperty( - HighAvailabilityOptions.HA_CLUSTER_ID, - haClusterId + "-restored"))); + HighAvailabilityOptions.HA_CLUSTER_ID, haClusterId))); arguments.addAll(unboundedJobArguments()); final Process second = startProcess("second", MiniClusterJobEntryPoint.class, arguments); @@ -1148,7 +1147,7 @@ class MiniClusterJobEntryPointITCase { jobId, expectedState, name, TIMEOUT, lastState.orElse("none")); } - private Optional<String> queryJobState() throws InterruptedException { + private Optional<String> queryJobState() throws Exception { return queryRest("/jobs/" + jobId).map(json -> json.get("state").asText()); } @@ -1210,9 +1209,34 @@ class MiniClusterJobEntryPointITCase { } } - private Optional<JsonNode> queryRest(String path) throws InterruptedException { + /** + * Returns the REST address published by the cluster's leader. After a process is killed, this + * may briefly be its stale address until its ZooKeeper session expires; requests to it are + * refused and retried. + */ + private Optional<String> restUrl() throws Exception { + final String path = + "/flink/" + + haClusterId + + ZooKeeperUtils.generateConnectionInformationPath( + ZooKeeperUtils.getLeaderPath(ZooKeeperUtils.getRestServerNode())); + try { + final LeaderInformation leader = + ZooKeeperUtils.readLeaderInformation(zooKeeperClient().getData().forPath(path)); + return leader.isEmpty() ? Optional.empty() : Optional.of(leader.getLeaderAddress()); + } catch (KeeperException e) { + // not yet published, or ZooKeeper is briefly unavailable + return Optional.empty(); + } + } + + private Optional<JsonNode> queryRest(String path) throws Exception { + final Optional<String> url = restUrl(); + if (url.isEmpty()) { + return Optional.empty(); + } final HttpRequest request = - HttpRequest.newBuilder(URI.create("http://localhost:"; + restPort + path)) + HttpRequest.newBuilder(URI.create(url.get() + path)) .timeout(REQUEST_TIMEOUT) .GET() .build(); @@ -1224,14 +1248,16 @@ class MiniClusterJobEntryPointITCase { } return Optional.of(OBJECT_MAPPER.readTree(response.body())); } catch (IOException e) { - // REST endpoint not up yet + // endpoint is busy, killed, or shutting down return Optional.empty(); } } private JsonNode postRest(String path, JsonNode body) throws Exception { + final String url = + restUrl().orElseThrow(() -> new IllegalStateException("No REST leader published.")); final HttpRequest request = - HttpRequest.newBuilder(URI.create("http://localhost:"; + restPort + path)) + HttpRequest.newBuilder(URI.create(url + path)) .timeout(REQUEST_TIMEOUT) .header("Content-Type", "application/json") .POST(HttpRequest.BodyPublishers.ofString(body.toString())) @@ -1295,13 +1321,14 @@ class MiniClusterJobEntryPointITCase { } private boolean zooKeeperNodeExists(String path) throws Exception { + return zooKeeperClient().checkExists().forPath(path) != null; + } + + private static CuratorFramework zooKeeperClient() { return ZOOKEEPER - .getCustomExtension() - .getZooKeeperClient(new TestingFatalErrorHandler()) - .usingNamespace(null) - .checkExists() - .forPath(path) - != null; + .getCustomExtension() + .getZooKeeperClient(new TestingFatalErrorHandler()) + .usingNamespace(null); } private String executionPlanZooKeeperPath() {
