You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/05/25 08:30:49 UTC
[flink] 06/08: [FLINK-32161][test] Migrate and remove legacy BlobServerResource.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c3409dcd0f77ef3343145ef1a05944a27f4f87ec
Author: Weijie Guo <re...@163.com>
AuthorDate: Tue May 23 17:31:33 2023 +0800
[FLINK-32161][test] Migrate and remove legacy BlobServerResource.
---
.../runner/DefaultDispatcherRunnerITCase.java | 139 +++++++++++----------
.../flink/runtime/util/BlobServerResource.java | 65 ----------
2 files changed, 70 insertions(+), 134 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
index 2c25c7e22bf..ce0dd9cd55b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.dispatcher.runner;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.dispatcher.Dispatcher;
import org.apache.flink.runtime.dispatcher.DispatcherBootstrapFactory;
import org.apache.flink.runtime.dispatcher.DispatcherFactory;
@@ -47,21 +48,20 @@ import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.jobmaster.TestingJobManagerRunner;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
import org.apache.flink.runtime.testutils.TestingJobGraphStore;
-import org.apache.flink.runtime.util.BlobServerResource;
+import org.apache.flink.runtime.util.BlobServerExtension;
import org.apache.flink.runtime.util.LeaderConnectionInfo;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,27 +71,26 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for the {@link DefaultDispatcherRunner}. */
-public class DefaultDispatcherRunnerITCase extends TestLogger {
+class DefaultDispatcherRunnerITCase {
private static final Logger LOG = LoggerFactory.getLogger(DefaultDispatcherRunnerITCase.class);
private static final Time TIMEOUT = Time.seconds(10L);
- @ClassRule
- public static TestingRpcServiceResource rpcServiceResource = new TestingRpcServiceResource();
+ @RegisterExtension
+ public static AllCallbackWrapper<TestingRpcServiceExtension> rpcServiceExtensionWrapper =
+ new AllCallbackWrapper<>(new TestingRpcServiceExtension());
- @ClassRule public static BlobServerResource blobServerResource = new BlobServerResource();
+ @RegisterExtension
+ public static AllCallbackWrapper<BlobServerExtension> blobServerExtensionWrapper =
+ new AllCallbackWrapper<>(new BlobServerExtension());
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ public static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
private JobGraph jobGraph;
@@ -107,8 +106,8 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
private DefaultDispatcherRunnerFactory dispatcherRunnerFactory;
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
dispatcherRunnerFactory =
DefaultDispatcherRunnerFactory.createSessionRunner(
SessionDispatcherFactory.INSTANCE);
@@ -121,18 +120,20 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
partialDispatcherServices =
TestingPartialDispatcherServices.builder()
.withFatalErrorHandler(fatalErrorHandler)
- .build(blobServerResource.getBlobServer(), new Configuration());
+ .build(
+ blobServerExtensionWrapper.getCustomExtension().getBlobServer(),
+ new Configuration());
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ void teardown() throws Exception {
if (fatalErrorHandler != null) {
fatalErrorHandler.rethrowError();
}
}
@Test
- public void leaderChange_afterJobSubmission_recoversSubmittedJob() throws Exception {
+ void leaderChange_afterJobSubmission_recoversSubmittedJob() throws Exception {
try (final DispatcherRunner dispatcherRunner = createDispatcherRunner()) {
final UUID firstLeaderSessionId = UUID.randomUUID();
@@ -149,7 +150,7 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
final Collection<JobID> jobIds = secondDispatcherGateway.listJobs(TIMEOUT).get();
- assertThat(jobIds, contains(jobGraph.getJobID()));
+ assertThat(jobIds).containsExactly(jobGraph.getJobID());
}
}
@@ -159,7 +160,8 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
final LeaderConnectionInfo leaderConnectionInfo =
dispatcherLeaderElectionService.getConfirmationFuture().get();
- return rpcServiceResource
+ return rpcServiceExtensionWrapper
+ .getCustomExtension()
.getTestingRpcService()
.connect(
leaderConnectionInfo.getAddress(),
@@ -173,8 +175,7 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
* fail.
*/
@Test
- public void leaderChange_withBlockingJobManagerTermination_doesNotAffectNewLeader()
- throws Exception {
+ void leaderChange_withBlockingJobManagerTermination_doesNotAffectNewLeader() throws Exception {
final TestingJobMasterServiceLeadershipRunnerFactory jobManagerRunnerFactory =
new TestingJobMasterServiceLeadershipRunnerFactory(1);
final TestingCleanupRunnerFactory cleanupRunnerFactory = new TestingCleanupRunnerFactory();
@@ -188,44 +189,44 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
// initial run
dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get();
- final TestingJobManagerRunner testingJobManagerRunner =
- jobManagerRunnerFactory.takeCreatedJobManagerRunner();
-
- dispatcherLeaderElectionService.notLeader();
-
- LOG.info("Re-grant leadership first time.");
- dispatcherLeaderElectionService.isLeader(UUID.randomUUID());
-
- // give the Dispatcher some time to recover jobs
- Thread.sleep(1L);
-
- dispatcherLeaderElectionService.notLeader();
-
- LOG.info("Re-grant leadership second time.");
- final UUID leaderSessionId = UUID.randomUUID();
- final CompletableFuture<UUID> leaderFuture =
- dispatcherLeaderElectionService.isLeader(leaderSessionId);
- assertThat(leaderFuture.isDone(), is(false));
-
- LOG.info("Complete the termination of the first job manager runner.");
- testingJobManagerRunner.completeTerminationFuture();
-
- assertThat(
- leaderFuture.get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS),
- is(equalTo(leaderSessionId)));
-
- // Wait for job to recover...
- final DispatcherGateway leaderGateway =
- rpcServiceResource
- .getTestingRpcService()
- .connect(
- dispatcherLeaderElectionService.getAddress(),
- DispatcherId.fromUuid(leaderSessionId),
- DispatcherGateway.class)
- .get();
- assertEquals(
- jobGraph.getJobID(),
- Iterables.getOnlyElement(leaderGateway.listJobs(TIMEOUT).get()));
+ try (final TestingJobManagerRunner testingJobManagerRunner =
+ jobManagerRunnerFactory.takeCreatedJobManagerRunner()) {
+
+ dispatcherLeaderElectionService.notLeader();
+
+ LOG.info("Re-grant leadership first time.");
+ dispatcherLeaderElectionService.isLeader(UUID.randomUUID());
+
+ // give the Dispatcher some time to recover jobs
+ Thread.sleep(1L);
+
+ dispatcherLeaderElectionService.notLeader();
+
+ LOG.info("Re-grant leadership second time.");
+ final UUID leaderSessionId = UUID.randomUUID();
+ final CompletableFuture<UUID> leaderFuture =
+ dispatcherLeaderElectionService.isLeader(leaderSessionId);
+ assertThat(leaderFuture).isNotDone();
+
+ LOG.info("Complete the termination of the first job manager runner.");
+ testingJobManagerRunner.completeTerminationFuture();
+
+ assertThat(leaderFuture.get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS))
+ .isEqualTo(leaderSessionId);
+
+ // Wait for job to recover...
+ final DispatcherGateway leaderGateway =
+ rpcServiceExtensionWrapper
+ .getCustomExtension()
+ .getTestingRpcService()
+ .connect(
+ dispatcherLeaderElectionService.getAddress(),
+ DispatcherId.fromUuid(leaderSessionId),
+ DispatcherGateway.class)
+ .get();
+ assertThat(Iterables.getOnlyElement(leaderGateway.listJobs(TIMEOUT).get()))
+ .isEqualTo(jobGraph.getJobID());
+ }
}
}
@@ -273,7 +274,7 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
fatalErrorHandler,
new TestingJobPersistenceComponentFactory(jobGraphStore, jobResultStore),
EXECUTOR_RESOURCE.getExecutor(),
- rpcServiceResource.getTestingRpcService(),
+ rpcServiceExtensionWrapper.getCustomExtension().getTestingRpcService(),
partialDispatcherServices);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlobServerResource.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlobServerResource.java
deleted file mode 100644
index 8e6910e3145..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlobServerResource.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.util;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.blob.BlobServer;
-import org.apache.flink.runtime.blob.VoidBlobStore;
-
-import org.junit.rules.ExternalResource;
-import org.junit.rules.TemporaryFolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-/** A simple {@link ExternalResource} to be used by tests that require a {@link BlobServer}. */
-public class BlobServerResource extends ExternalResource {
- private static final Logger LOG = LoggerFactory.getLogger(BlobServerResource.class);
- private final TemporaryFolder temporaryFolder = new TemporaryFolder();
-
- private BlobServer blobServer;
-
- protected void before() throws Throwable {
- temporaryFolder.create();
-
- Configuration config = new Configuration();
-
- blobServer = new BlobServer(config, temporaryFolder.newFolder(), new VoidBlobStore());
- blobServer.start();
- }
-
- protected void after() {
- temporaryFolder.delete();
-
- try {
- blobServer.close();
- } catch (IOException e) {
- LOG.error("Exception while shutting down blob server.", e);
- }
- }
-
- public int getBlobServerPort() {
- return blobServer.getPort();
- }
-
- public BlobServer getBlobServer() {
- return blobServer;
- }
-}