You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/18 09:13:54 UTC
[1/9] flink git commit: [hotfix] [yarn] Write number of slots to
configuration
Repository: flink
Updated Branches:
refs/heads/master 9b7e42924 -> 3a4e89643
[hotfix] [yarn] Write number of slots to configuration
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4b89b5d0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4b89b5d0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4b89b5d0
Branch: refs/heads/master
Commit: 4b89b5d0abe4684dbeacd9a01fe3297621050eb7
Parents: 075f5b6
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Feb 7 18:21:06 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:52 2018 +0100
----------------------------------------------------------------------
.../org/apache/flink/yarn/AbstractYarnClusterDescriptor.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4b89b5d0/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index 583ee54..d3fce4c 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -482,7 +482,7 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
flinkConfiguration.setString(ClusterEntrypoint.EXECUTION_MODE, executionMode.toString());
ApplicationReport report = startAppMaster(
- flinkConfiguration,
+ new Configuration(flinkConfiguration),
yarnClusterEntrypoint,
jobGraph,
yarnClient,
@@ -794,6 +794,10 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
homeDir,
"");
+ configuration.setInteger(
+ ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS,
+ clusterSpecification.getSlotsPerTaskManager());
+
// Upload the flink configuration
// write out configuration file
File tmpConfigurationFile = File.createTempFile(appId + "-flink-conf.yaml", null);
[5/9] flink git commit: [FLINK-8653] [flip6] Remove internal slot
request timeout from SlotPool
Posted by tr...@apache.org.
[FLINK-8653] [flip6] Remove internal slot request timeout from SlotPool
Instead of using the internal slot request timeout to time out pending slot requests,
we use the timeout passed to SlotPool#allocateSlot to time out pending slot requests.
This closes #5483.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d8a88669
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d8a88669
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d8a88669
Branch: refs/heads/master
Commit: d8a8866973f7e0463047963e6b242cdc2cb82fec
Parents: d9d89ff
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Feb 14 12:09:01 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:54 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/jobmaster/JobMaster.java | 1 -
.../runtime/jobmaster/slotpool/SlotPool.java | 89 ++++++++++----------
.../jobmaster/slotpool/SlotPoolRpcTest.java | 72 ++++++----------
.../jobmaster/slotpool/SlotPoolTest.java | 34 +-------
4 files changed, 74 insertions(+), 122 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d8a88669/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 139c053..dfa4d1c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -276,7 +276,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
jobGraph.getJobID(),
SystemClock.getInstance(),
rpcTimeout,
- jobMasterConfiguration.getSlotRequestTimeout(),
jobMasterConfiguration.getSlotIdleTimeout());
this.slotPoolGateway = slotPool.getSelfGateway(SlotPoolGateway.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/d8a88669/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index 72bb7e1..6d714e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -67,6 +67,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -111,9 +112,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
/** Timeout for external request calls (e.g. to the ResourceManager or the TaskExecutor). */
private final Time rpcTimeout;
- /** Timeout for allocation round trips (RM -> launch TM -> offer slot). */
- private final Time slotRequestTimeout;
-
/** Timeout for releasing idle slots. */
private final Time idleSlotTimeout;
@@ -139,7 +137,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
jobId,
SystemClock.getInstance(),
AkkaUtils.getDefaultTimeout(),
- Time.milliseconds(JobManagerOptions.SLOT_REQUEST_TIMEOUT.defaultValue()),
Time.milliseconds(JobManagerOptions.SLOT_IDLE_TIMEOUT.defaultValue()));
}
@@ -148,7 +145,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
JobID jobId,
Clock clock,
Time rpcTimeout,
- Time slotRequestTimeout,
Time idleSlotTimeout) {
super(rpcService);
@@ -157,7 +153,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
this.clock = checkNotNull(clock);
this.rpcTimeout = checkNotNull(rpcTimeout);
this.idleSlotTimeout = checkNotNull(idleSlotTimeout);
- this.slotRequestTimeout = checkNotNull(slotRequestTimeout);
this.registeredTaskManagers = new HashSet<>(16);
this.allocatedSlots = new AllocatedSlots();
@@ -307,7 +302,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
scheduledUnit,
resourceProfile,
locationPreferences,
- allowQueuedScheduling);
+ allowQueuedScheduling,
+ timeout);
}
private CompletableFuture<LogicalSlot> internalAllocateSlot(
@@ -315,7 +311,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
ScheduledUnit task,
ResourceProfile resourceProfile,
Collection<TaskManagerLocation> locationPreferences,
- boolean allowQueuedScheduling) {
+ boolean allowQueuedScheduling,
+ Time allocationTimeout) {
final SlotSharingGroupId slotSharingGroupId = task.getSlotSharingGroupId();
@@ -337,13 +334,15 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
multiTaskSlotManager,
resourceProfile,
locationPreferences,
- allowQueuedScheduling);
+ allowQueuedScheduling,
+ allocationTimeout);
} else {
multiTaskSlotLocality = allocateMultiTaskSlot(
task.getJobVertexId(), multiTaskSlotManager,
resourceProfile,
locationPreferences,
- allowQueuedScheduling);
+ allowQueuedScheduling,
+ allocationTimeout);
}
} catch (NoResourceAvailableException noResourceException) {
return FutureUtils.completedExceptionally(noResourceException);
@@ -364,7 +363,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
slotRequestId,
resourceProfile,
locationPreferences,
- allowQueuedScheduling);
+ allowQueuedScheduling,
+ allocationTimeout);
return slotAndLocalityFuture.thenApply(
(SlotAndLocality slotAndLocality) -> {
@@ -399,6 +399,7 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
* @param resourceProfile specifying the requirements for the requested slot
* @param locationPreferences containing preferred TaskExecutors on which to allocate the slot
* @param allowQueuedScheduling true if queued scheduling (the returned task slot must not be completed yet) is allowed, otherwise false
+ * @param allocationTimeout timeout before the slot allocation times out
* @return A {@link SlotSharingManager.MultiTaskSlotLocality} which contains the allocated{@link SlotSharingManager.MultiTaskSlot}
* and its locality wrt the given location preferences
* @throws NoResourceAvailableException if no task slot could be allocated
@@ -408,7 +409,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
SlotSharingManager multiTaskSlotManager,
ResourceProfile resourceProfile,
Collection<TaskManagerLocation> locationPreferences,
- boolean allowQueuedScheduling) throws NoResourceAvailableException {
+ boolean allowQueuedScheduling,
+ Time allocationTimeout) throws NoResourceAvailableException {
final SlotRequestId coLocationSlotRequestId = coLocationConstraint.getSlotRequestId();
if (coLocationSlotRequestId != null) {
@@ -437,7 +439,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
coLocationConstraint.getGroupId(), multiTaskSlotManager,
resourceProfile,
actualLocationPreferences,
- allowQueuedScheduling);
+ allowQueuedScheduling,
+ allocationTimeout);
// check whether we fulfill the co-location constraint
if (coLocationConstraint.isAssigned() && multiTaskSlotLocality.getLocality() != Locality.LOCAL) {
@@ -493,6 +496,7 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
* @param resourceProfile specifying the requirements for the requested slot
* @param locationPreferences containing preferred TaskExecutors on which to allocate the slot
* @param allowQueuedScheduling true if queued scheduling (the returned task slot must not be completed yet) is allowed, otherwise false
+ * @param allocationTimeout timeout before the slot allocation times out
* @return A {@link SlotSharingManager.MultiTaskSlotLocality} which contains the allocated {@link SlotSharingManager.MultiTaskSlot}
* and its locality wrt the given location preferences
* @throws NoResourceAvailableException if no task slot could be allocated
@@ -502,7 +506,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
SlotSharingManager slotSharingManager,
ResourceProfile resourceProfile,
Collection<TaskManagerLocation> locationPreferences,
- boolean allowQueuedScheduling) throws NoResourceAvailableException {
+ boolean allowQueuedScheduling,
+ Time allocationTimeout) throws NoResourceAvailableException {
// check first whether we have a resolved root slot which we can use
SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality = slotSharingManager.getResolvedRootSlot(
@@ -552,7 +557,10 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
if (multiTaskSlotFuture == null) {
// it seems as if we have to request a new slot from the resource manager, this is always the last resort!!!
- final CompletableFuture<AllocatedSlot> futureSlot = requestNewAllocatedSlot(allocatedSlotRequestId, resourceProfile);
+ final CompletableFuture<AllocatedSlot> futureSlot = requestNewAllocatedSlot(
+ allocatedSlotRequestId,
+ resourceProfile,
+ allocationTimeout);
multiTaskSlotFuture = slotSharingManager.createRootSlot(
multiTaskSlotRequestId,
@@ -597,13 +605,15 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
* @param resourceProfile which the allocated slot should fulfill
* @param locationPreferences for the allocated slot
* @param allowQueuedScheduling true if the slot allocation can be completed in the future
+ * @param allocationTimeout timeout before the slot allocation times out
* @return Future containing the allocated simple slot
*/
private CompletableFuture<SlotAndLocality> requestAllocatedSlot(
SlotRequestId slotRequestId,
ResourceProfile resourceProfile,
Collection<TaskManagerLocation> locationPreferences,
- boolean allowQueuedScheduling) {
+ boolean allowQueuedScheduling,
+ Time allocationTimeout) {
final CompletableFuture<SlotAndLocality> allocatedSlotLocalityFuture;
@@ -616,7 +626,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
// we have to request a new allocated slot
CompletableFuture<AllocatedSlot> allocatedSlotFuture = requestNewAllocatedSlot(
slotRequestId,
- resourceProfile);
+ resourceProfile,
+ allocationTimeout);
allocatedSlotLocalityFuture = allocatedSlotFuture.thenApply((AllocatedSlot allocatedSlot) -> new SlotAndLocality(allocatedSlot, Locality.UNKNOWN));
} else {
@@ -634,16 +645,29 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
*
* @param slotRequestId identifying the requested slot
* @param resourceProfile which the requested slot should fulfill
+ * @param allocationTimeout timeout before the slot allocation times out
* @return An {@link AllocatedSlot} future which is completed once the slot is offered to the {@link SlotPool}
*/
private CompletableFuture<AllocatedSlot> requestNewAllocatedSlot(
- SlotRequestId slotRequestId,
- ResourceProfile resourceProfile) {
+ SlotRequestId slotRequestId,
+ ResourceProfile resourceProfile,
+ Time allocationTimeout) {
final PendingRequest pendingRequest = new PendingRequest(
slotRequestId,
resourceProfile);
+ // register request timeout
+ FutureUtils
+ .orTimeout(pendingRequest.getAllocatedSlotFuture(), allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS)
+ .whenCompleteAsync(
+ (AllocatedSlot ignored, Throwable throwable) -> {
+ if (throwable != null) {
+ removePendingRequest(slotRequestId);
+ }
+ },
+ getMainThreadExecutor());
+
if (resourceManagerGateway == null) {
stashRequestWaitingForResourceManager(pendingRequest);
} else {
@@ -678,15 +702,9 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
new SlotRequest(jobId, allocationId, pendingRequest.getResourceProfile(), jobManagerAddress),
rpcTimeout);
- CompletableFuture<Void> slotRequestProcessingFuture = rmResponse.thenAcceptAsync(
- (Acknowledge value) -> {
- slotRequestToResourceManagerSuccess(pendingRequest.getSlotRequestId());
- },
- getMainThreadExecutor());
-
// on failure, fail the request future
- slotRequestProcessingFuture.whenCompleteAsync(
- (Void v, Throwable failure) -> {
+ rmResponse.whenCompleteAsync(
+ (Acknowledge ignored, Throwable failure) -> {
if (failure != null) {
slotRequestToResourceManagerFailed(pendingRequest.getSlotRequestId(), failure);
}
@@ -694,12 +712,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
getMainThreadExecutor());
}
- private void slotRequestToResourceManagerSuccess(final SlotRequestId requestId) {
- // a request is pending from the ResourceManager to a (future) TaskManager
- // we only add the watcher here in case that request times out
- scheduleRunAsync(() -> checkTimeoutSlotAllocation(requestId), slotRequestTimeout);
- }
-
private void slotRequestToResourceManagerFailed(SlotRequestId slotRequestID, Throwable failure) {
PendingRequest request = pendingRequests.removeKeyA(slotRequestID);
if (request != null) {
@@ -727,17 +739,6 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
"Adding as pending request {}", pendingRequest.getSlotRequestId());
waitingForResourceManager.put(pendingRequest.getSlotRequestId(), pendingRequest);
-
- scheduleRunAsync(() -> checkTimeoutRequestWaitingForResourceManager(pendingRequest.getSlotRequestId()), slotRequestTimeout);
- }
-
- private void checkTimeoutRequestWaitingForResourceManager(SlotRequestId slotRequestId) {
- PendingRequest request = waitingForResourceManager.remove(slotRequestId);
- if (request != null) {
- failPendingRequest(
- request,
- new TimeoutException("No slot available and no connection to Resource Manager established."));
- }
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d8a88669/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRpcTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRpcTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRpcTest.java
index bf8037d..a9be9cf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRpcTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRpcTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.runtime.jobmanager.scheduler.DummyScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
@@ -50,7 +51,6 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import akka.actor.ActorSystem;
-import akka.pattern.AskTimeoutException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -66,7 +66,6 @@ import java.util.function.Consumer;
import static org.apache.flink.runtime.jobmaster.slotpool.AvailableSlotsTest.DEFAULT_TESTING_PROFILE;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -80,6 +79,8 @@ public class SlotPoolRpcTest extends TestLogger {
private static final Time timeout = Time.seconds(10L);
+ private static final Time fastTimeout = Time.milliseconds(1L);
+
// ------------------------------------------------------------------------
// setup
// ------------------------------------------------------------------------
@@ -111,7 +112,6 @@ public class SlotPoolRpcTest extends TestLogger {
jid,
SystemClock.getInstance(),
TestingUtils.infiniteTime(),
- Time.milliseconds(10L), // this is the timeout for the request tested here
TestingUtils.infiniteTime()
);
@@ -124,7 +124,7 @@ public class SlotPoolRpcTest extends TestLogger {
DEFAULT_TESTING_PROFILE,
Collections.emptyList(),
true,
- TestingUtils.infiniteTime());
+ fastTimeout);
try {
future.get();
@@ -146,7 +146,6 @@ public class SlotPoolRpcTest extends TestLogger {
jid,
SystemClock.getInstance(),
TestingUtils.infiniteTime(),
- TestingUtils.infiniteTime(),
TestingUtils.infiniteTime());
try {
@@ -160,27 +159,26 @@ public class SlotPoolRpcTest extends TestLogger {
DEFAULT_TESTING_PROFILE,
Collections.emptyList(),
true,
- Time.milliseconds(10L));
+ fastTimeout);
try {
future.get();
- fail("We expected a AskTimeoutException.");
+ fail("We expected a TimeoutException.");
} catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e) instanceof AskTimeoutException);
+ assertTrue(ExceptionUtils.stripExecutionException(e) instanceof TimeoutException);
}
- assertEquals(1L, (long) pool.getNumberOfWaitingForResourceRequests().get());
-
- slotPoolGateway.releaseSlot(requestId, null, null).get();
-
assertEquals(0L, (long) pool.getNumberOfWaitingForResourceRequests().get());
} finally {
RpcUtils.terminateRpcEndpoint(pool, timeout);
}
}
+ /**
+ * Tests that a slot allocation times out wrt to the specified time out.
+ */
@Test
- public void testCancelSlotAllocationWithResourceManager() throws Exception {
+ public void testSlotAllocationTimeout() throws Exception {
final JobID jid = new JobID();
final TestingSlotPool pool = new TestingSlotPool(
@@ -188,7 +186,6 @@ public class SlotPoolRpcTest extends TestLogger {
jid,
SystemClock.getInstance(),
TestingUtils.infiniteTime(),
- TestingUtils.infiniteTime(),
TestingUtils.infiniteTime());
try {
@@ -201,22 +198,19 @@ public class SlotPoolRpcTest extends TestLogger {
SlotRequestId requestId = new SlotRequestId();
CompletableFuture<LogicalSlot> future = slotPoolGateway.allocateSlot(
requestId,
- new ScheduledUnit(SchedulerTestUtils.getDummyTask()),
+ new DummyScheduledUnit(),
DEFAULT_TESTING_PROFILE,
Collections.emptyList(),
true,
- Time.milliseconds(10L));
+ fastTimeout);
try {
future.get();
- fail("We expected a AskTimeoutException.");
+ fail("We expected a TimeoutException.");
} catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e) instanceof AskTimeoutException);
+ assertTrue(ExceptionUtils.stripExecutionException(e) instanceof TimeoutException);
}
- assertEquals(1L, (long) pool.getNumberOfPendingRequests().get());
-
- slotPoolGateway.releaseSlot(requestId, null, null).get();
assertEquals(0L, (long) pool.getNumberOfPendingRequests().get());
} finally {
RpcUtils.terminateRpcEndpoint(pool, timeout);
@@ -224,10 +218,10 @@ public class SlotPoolRpcTest extends TestLogger {
}
/**
- * Tests that allocated slots are not cancelled.
+ * Tests that extra slots are kept by the {@link SlotPool}.
*/
@Test
- public void testCancelSlotAllocationWhileSlotFulfilled() throws Exception {
+ public void testExtraSlotsAreKept() throws Exception {
final JobID jid = new JobID();
final TestingSlotPool pool = new TestingSlotPool(
@@ -235,7 +229,6 @@ public class SlotPoolRpcTest extends TestLogger {
jid,
SystemClock.getInstance(),
TestingUtils.infiniteTime(),
- TestingUtils.infiniteTime(),
TestingUtils.infiniteTime());
try {
@@ -257,15 +250,17 @@ public class SlotPoolRpcTest extends TestLogger {
DEFAULT_TESTING_PROFILE,
Collections.emptyList(),
true,
- Time.milliseconds(10L));
+ fastTimeout);
try {
future.get();
- fail("We expected a AskTimeoutException.");
+ fail("We expected a TimeoutException.");
} catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e) instanceof AskTimeoutException);
+ assertTrue(ExceptionUtils.stripExecutionException(e) instanceof TimeoutException);
}
+ assertEquals(0L, (long) pool.getNumberOfPendingRequests().get());
+
AllocationID allocationId = allocationIdFuture.get();
final SlotOffer slotOffer = new SlotOffer(
allocationId,
@@ -278,13 +273,6 @@ public class SlotPoolRpcTest extends TestLogger {
assertTrue(slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, slotOffer).get());
- assertEquals(0L, (long) pool.getNumberOfPendingRequests().get());
-
- assertTrue(pool.containsAllocatedSlot(allocationId).get());
-
- pool.releaseSlot(requestId, null, null).get();
-
- assertFalse(pool.containsAllocatedSlot(allocationId).get());
assertTrue(pool.containsAvailableSlot(allocationId).get());
} finally {
RpcUtils.terminateRpcEndpoint(pool, timeout);
@@ -296,7 +284,7 @@ public class SlotPoolRpcTest extends TestLogger {
* it will automatically call cancelSlotAllocation as will inject future.whenComplete in ProviderAndOwner.
*/
@Test
- public void testProviderAndOwner() throws Exception {
+ public void testProviderAndOwnerSlotAllocationTimeout() throws Exception {
final JobID jid = new JobID();
final TestingSlotPool pool = new TestingSlotPool(
@@ -304,7 +292,6 @@ public class SlotPoolRpcTest extends TestLogger {
jid,
SystemClock.getInstance(),
TestingUtils.infiniteTime(),
- TestingUtils.infiniteTime(),
TestingUtils.infiniteTime());
final CompletableFuture<SlotRequestId> releaseSlotFuture = new CompletableFuture<>();
@@ -317,20 +304,18 @@ public class SlotPoolRpcTest extends TestLogger {
ResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
pool.connectToResourceManager(resourceManagerGateway);
- ScheduledUnit mockScheduledUnit = new ScheduledUnit(SchedulerTestUtils.getDummyTask());
-
// test the pending request is clear when timed out
CompletableFuture<LogicalSlot> future = pool.getSlotProvider().allocateSlot(
- mockScheduledUnit,
+ new DummyScheduledUnit(),
true,
Collections.emptyList(),
- Time.milliseconds(1L));
+ fastTimeout);
try {
future.get();
- fail("We expected a AskTimeoutException.");
+ fail("We expected a TimeoutException.");
} catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e) instanceof AskTimeoutException);
+ assertTrue(ExceptionUtils.stripExecutionException(e) instanceof TimeoutException);
}
// wait for the cancel call on the SlotPool
@@ -353,14 +338,13 @@ public class SlotPoolRpcTest extends TestLogger {
RpcService rpcService,
JobID jobId,
Clock clock,
- Time slotRequestTimeout,
Time rpcTimeout,
Time idleSlotTimeout) {
super(
rpcService,
jobId,
clock,
- rpcTimeout, slotRequestTimeout,
+ rpcTimeout,
idleSlotTimeout);
releaseSlotConsumer = null;
http://git-wip-us.apache.org/repos/asf/flink/blob/d8a88669/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
index 6a85461..e6446ad 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
@@ -66,7 +66,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import static org.apache.flink.runtime.jobmaster.slotpool.AvailableSlotsTest.DEFAULT_TESTING_PROFILE;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -587,7 +586,7 @@ public class SlotPoolTest extends TestLogger {
rpcService,
jobId,
clock,
- TestingUtils.infiniteTime(), TestingUtils.infiniteTime(),
+ TestingUtils.infiniteTime(),
timeout);
try {
@@ -628,37 +627,6 @@ public class SlotPoolTest extends TestLogger {
}
}
- /**
- * Tests that a slot allocation times out wrt to the specified time out.
- */
- @Test
- public void testSlotAllocationTimeout() throws Exception {
- final SlotPool slotPool = new SlotPool(rpcService, jobId);
-
- final Time allocationTimeout = Time.milliseconds(1L);
-
- try {
- setupSlotPool(slotPool, resourceManagerGateway);
-
- final SlotProvider slotProvider = slotPool.getSlotProvider();
-
- final CompletableFuture<LogicalSlot> allocationFuture = slotProvider.allocateSlot(
- new DummyScheduledUnit(),
- true,
- Collections.emptyList(),
- allocationTimeout);
-
- try {
- allocationFuture.get();
- fail("Should have failed with a timeout exception.");
- } catch (ExecutionException ee) {
- assertThat(ExceptionUtils.stripExecutionException(ee), Matchers.instanceOf(TimeoutException.class));
- }
- } finally {
- RpcUtils.terminateRpcEndpoint(slotPool, timeout);
- }
- }
-
private static SlotPoolGateway setupSlotPool(
SlotPool slotPool,
ResourceManagerGateway resourceManagerGateway) throws Exception {
[2/9] flink git commit: [FLINK-8673] [flip6] Use
JobManagerRunner#resultFuture for success and failure communication
Posted by tr...@apache.org.
[FLINK-8673] [flip6] Use JobManagerRunner#resultFuture for success and failure communication
This commit removes the OnCompletionActions and FatalErrorHandler from the
JobManagerRunner. Instead it communicates a successful job execution of the
failure case through the JobManagerRunner#resultFuture.
Furthermore, this commit no longer allows the JobManagerRunner to shut down itself.
All shut down logic must be triggered by the owner of the JobManagerRunner.
This closes #5510.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/075f5b69
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/075f5b69
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/075f5b69
Branch: refs/heads/master
Commit: 075f5b6930b86e110a32e290229b9800be72a3a7
Parents: 9b7e429
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Feb 16 15:04:32 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:52 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/dispatcher/Dispatcher.java | 60 ++----
.../runtime/dispatcher/MiniDispatcher.java | 4 +-
.../runtime/jobmaster/JobManagerRunner.java | 86 +++-----
.../jobmaster/JobNotFinishedException.java | 33 +++
.../minicluster/MiniClusterJobDispatcher.java | 44 +++-
.../apache/flink/runtime/rpc/RpcService.java | 4 +-
.../runtime/dispatcher/DispatcherTest.java | 21 +-
.../runtime/dispatcher/MiniDispatcherTest.java | 55 +++--
.../runtime/jobmaster/JobManagerRunnerTest.java | 201 ++++++++++++++++++-
9 files changed, 370 insertions(+), 138 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 89b82ef..e751bc4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -34,11 +34,11 @@ import org.apache.flink.runtime.highavailability.RunningJobsRegistry;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
+import org.apache.flink.runtime.jobmaster.JobNotFinishedException;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.leaderelection.LeaderContender;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
@@ -243,10 +243,23 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
blobServer,
jobManagerSharedServices,
metricRegistry,
- new DispatcherOnCompleteActions(jobGraph.getJobID()),
- fatalErrorHandler,
restAddress);
+ jobManagerRunner.getResultFuture().whenCompleteAsync(
+ (ArchivedExecutionGraph archivedExecutionGraph, Throwable throwable) -> {
+ if (archivedExecutionGraph != null) {
+ jobReachedGloballyTerminalState(archivedExecutionGraph);
+ } else {
+ final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable);
+
+ if (strippedThrowable instanceof JobNotFinishedException) {
+ jobNotFinished(jobId);
+ } else {
+ onFatalError(new FlinkException("JobManagerRunner for job " + jobId + " failed.", strippedThrowable));
+ }
+ }
+ }, getMainThreadExecutor());
+
jobManagerRunner.start();
} catch (Exception e) {
try {
@@ -544,6 +557,8 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
archivedExecutionGraph.getJobID(),
archivedExecutionGraph.getState());
+ log.info("Job {} reached globally terminal state {}.", archivedExecutionGraph.getJobID(), archivedExecutionGraph.getState());
+
try {
archivedExecutionGraphStore.put(archivedExecutionGraph);
} catch (IOException e) {
@@ -563,7 +578,9 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
}
}
- protected void jobFinishedByOther(JobID jobId) {
+ protected void jobNotFinished(JobID jobId) {
+ log.info("Job {} was not finished by JobManager.", jobId);
+
try {
removeJob(jobId, false);
} catch (Exception e) {
@@ -668,35 +685,6 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
}
//------------------------------------------------------
- // Utility classes
- //------------------------------------------------------
-
- @VisibleForTesting
- class DispatcherOnCompleteActions implements OnCompletionActions {
-
- private final JobID jobId;
-
- DispatcherOnCompleteActions(JobID jobId) {
- this.jobId = Preconditions.checkNotNull(jobId);
- }
-
- @Override
- public void jobReachedGloballyTerminalState(ArchivedExecutionGraph executionGraph) {
- log.info("Job {} reached globally terminal state {}.", jobId, executionGraph.getState());
-
- runAsync(() -> Dispatcher.this.jobReachedGloballyTerminalState(executionGraph));
- }
-
- @Override
- public void jobFinishedByOther() {
- log.info("Job {} was finished by other JobManager.", jobId);
-
- runAsync(
- () -> Dispatcher.this.jobFinishedByOther(jobId));
- }
- }
-
- //------------------------------------------------------
// Factories
//------------------------------------------------------
@@ -715,8 +703,6 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
MetricRegistry metricRegistry,
- OnCompletionActions onCompleteActions,
- FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception;
}
@@ -737,8 +723,6 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
MetricRegistry metricRegistry,
- OnCompletionActions onCompleteActions,
- FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
return new JobManagerRunner(
resourceId,
@@ -750,8 +734,6 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
blobServer,
jobManagerServices,
metricRegistry,
- onCompleteActions,
- fatalErrorHandler,
restAddress);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
index 36732ce..c648131 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -124,8 +124,8 @@ public class MiniDispatcher extends Dispatcher {
}
@Override
- protected void jobFinishedByOther(JobID jobId) {
- super.jobFinishedByOther(jobId);
+ protected void jobNotFinished(JobID jobId) {
+ super.jobNotFinished(jobId);
// shut down since we have done our job
shutDown();
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
index 5e476bf..285cb4a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
@@ -51,7 +51,6 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -72,12 +71,6 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
/** The job graph needs to run. */
private final JobGraph jobGraph;
- /** The listener to notify once the job completes - either successfully or unsuccessfully. */
- private final OnCompletionActions toNotifyOnComplete;
-
- /** The handler to call in case of fatal (unrecoverable) errors. */
- private final FatalErrorHandler errorHandler;
-
/** Used to check whether a job needs to be run. */
private final RunningJobsRegistry runningJobsRegistry;
@@ -94,6 +87,8 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
private final CompletableFuture<ArchivedExecutionGraph> resultFuture;
+ private final CompletableFuture<Void> terminationFuture;
+
/** flag marking the runner as shut down. */
private volatile boolean shutdown;
@@ -119,17 +114,16 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
final BlobServer blobServer,
final JobManagerSharedServices jobManagerSharedServices,
final MetricRegistry metricRegistry,
- final OnCompletionActions toNotifyOnComplete,
- final FatalErrorHandler errorHandler,
@Nullable final String restAddress) throws Exception {
JobManagerMetricGroup jobManagerMetrics = null;
+ this.resultFuture = new CompletableFuture<>();
+ this.terminationFuture = new CompletableFuture<>();
+
// make sure we cleanly shut down out JobManager services if initialization fails
try {
this.jobGraph = checkNotNull(jobGraph);
- this.toNotifyOnComplete = checkNotNull(toNotifyOnComplete);
- this.errorHandler = checkNotNull(errorHandler);
this.jobManagerSharedServices = checkNotNull(jobManagerSharedServices);
checkArgument(jobGraph.getNumberOfVertices() > 0, "The given job is empty");
@@ -176,8 +170,6 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
userCodeLoader,
restAddress,
metricRegistry.getMetricQueryServicePath());
-
- this.resultFuture = new CompletableFuture<>();
}
catch (Throwable t) {
// clean up everything
@@ -185,6 +177,9 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
jobManagerMetrics.close();
}
+ terminationFuture.completeExceptionally(t);
+ resultFuture.completeExceptionally(t);
+
throw new JobExecutionException(jobGraph.getJobID(), "Could not set up JobManager", t);
}
}
@@ -225,34 +220,43 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
private CompletableFuture<Void> shutdownInternally() {
synchronized (lock) {
- shutdown = true;
+ if (!shutdown) {
+ shutdown = true;
- jobManager.shutDown();
+ jobManager.shutDown();
- return jobManager.getTerminationFuture()
- .thenAccept(
- ignored -> {
- Throwable exception = null;
+ final CompletableFuture<Boolean> jobManagerTerminationFuture = jobManager.getTerminationFuture();
+
+ jobManagerTerminationFuture.whenComplete(
+ (Boolean ignored, Throwable throwable) -> {
try {
leaderElectionService.stop();
} catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ throwable = ExceptionUtils.firstOrSuppressed(t, ExceptionUtils.stripCompletionException(throwable));
}
// make all registered metrics go away
try {
jobManagerMetricGroup.close();
} catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ throwable = ExceptionUtils.firstOrSuppressed(t, throwable);
}
- if (exception != null) {
- throw new CompletionException(new FlinkException("Could not properly shut down the JobManagerRunner.", exception));
+ if (throwable != null) {
+ terminationFuture.completeExceptionally(
+ new FlinkException("Could not properly shut down the JobManagerRunner", throwable));
+ } else {
+ terminationFuture.complete(null);
}
+ });
- // cancel the result future if not already completed
- resultFuture.cancel(false);
+ terminationFuture.whenComplete(
+ (Void ignored, Throwable throwable) -> {
+ resultFuture.completeExceptionally(new JobNotFinishedException(jobGraph.getJobID()));
});
+ }
+
+ return terminationFuture;
}
}
@@ -267,16 +271,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
public void jobReachedGloballyTerminalState(ArchivedExecutionGraph executionGraph) {
// complete the result future with the terminal execution graph
resultFuture.complete(executionGraph);
-
- try {
- unregisterJobFromHighAvailability();
- shutdownInternally();
- }
- finally {
- if (toNotifyOnComplete != null) {
- toNotifyOnComplete.jobReachedGloballyTerminalState(executionGraph);
- }
- }
+ unregisterJobFromHighAvailability();
}
/**
@@ -284,14 +279,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
*/
@Override
public void jobFinishedByOther() {
- try {
- shutdownInternally();
- }
- finally {
- if (toNotifyOnComplete != null) {
- toNotifyOnComplete.jobFinishedByOther();
- }
- }
+ resultFuture.completeExceptionally(new JobNotFinishedException(jobGraph.getJobID()));
}
/**
@@ -306,17 +294,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
log.error("JobManager runner encountered a fatal error.", exception);
} catch (Throwable ignored) {}
- // in any case, notify our handler, so it can react fast
- try {
- if (errorHandler != null) {
- errorHandler.onFatalError(exception);
- }
- }
- finally {
- // the shutdown may not even needed any more, if the fatal error
- // handler kills the process. that is fine, a process kill cleans up better than anything.
- shutdownInternally();
- }
+ resultFuture.completeExceptionally(exception);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobNotFinishedException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobNotFinishedException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobNotFinishedException.java
new file mode 100644
index 0000000..3515f2f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobNotFinishedException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jobmaster;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.JobException;
+
+/**
+ * Exception indicating that a Flink job has not been finished.
+ */
+public class JobNotFinishedException extends JobException {
+ private static final long serialVersionUID = 611413276562570622L;
+
+ public JobNotFinishedException(JobID jobId) {
+ super("The job (" + jobId + ") has been not been finished.");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
index 6fba534..ef7a6e4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
@@ -32,6 +32,7 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
+import org.apache.flink.runtime.jobmaster.JobNotFinishedException;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
@@ -142,10 +143,10 @@ public class MiniClusterJobDispatcher {
MetricRegistry metricRegistry,
int numJobManagers,
RpcService[] rpcServices) throws Exception {
-
+
checkArgument(numJobManagers >= 1);
checkArgument(rpcServices.length == numJobManagers);
-
+
this.configuration = checkNotNull(config);
this.rpcServices = rpcServices;
this.haServices = checkNotNull(haServices);
@@ -237,7 +238,7 @@ public class MiniClusterJobDispatcher {
* This method runs a job in blocking mode. The method returns only after the job
* completed successfully, or after it failed terminally.
*
- * @param job The Flink job to execute
+ * @param job The Flink job to execute
* @return The result of the job execution
*
* @throws JobExecutionException Thrown if anything went amiss during initial job launch,
@@ -245,7 +246,7 @@ public class MiniClusterJobDispatcher {
*/
public JobExecutionResult runJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
checkNotNull(job);
-
+
LOG.info("Received job for blocking execution: {} ({})", job.getName(), job.getJobID());
final BlockingJobSync sync = new BlockingJobSync(job.getJobID(), numJobManagers);
@@ -287,9 +288,32 @@ public class MiniClusterJobDispatcher {
blobServer,
jobManagerSharedServices,
metricRegistry,
- onCompletion,
- errorHandler,
null);
+
+ final int index = i;
+
+ runners[i].getResultFuture()
+ .whenComplete(
+ (ArchivedExecutionGraph archivedExecutionGraph, Throwable throwable) -> {
+ try {
+ runners[index].shutdown();
+ } catch (Exception e) {
+ errorHandler.onFatalError(e);
+ }
+
+ if (archivedExecutionGraph != null) {
+ onCompletion.jobReachedGloballyTerminalState(archivedExecutionGraph);
+ } else {
+ final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable);
+
+ if (strippedThrowable instanceof JobNotFinishedException) {
+ onCompletion.jobFinishedByOther();
+ } else {
+ errorHandler.onFatalError(strippedThrowable);
+ }
+ }
+ });
+
runners[i].start();
}
catch (Throwable t) {
@@ -395,8 +419,8 @@ public class MiniClusterJobDispatcher {
* This class is used to sync on blocking jobs across multiple runners.
* Only after all runners reported back that they are finished, the
* result will be released.
- *
- * That way it is guaranteed that after the blocking job submit call returns,
+ *
+ * <p>That way it is guaranteed that after the blocking job submit call returns,
* the dispatcher is immediately free to accept another job.
*/
private static class BlockingJobSync implements OnCompletionActions, FatalErrorHandler {
@@ -408,7 +432,7 @@ public class MiniClusterJobDispatcher {
private volatile Throwable runnerException;
private volatile JobResult result;
-
+
BlockingJobSync(JobID jobId, int numJobMastersToWaitFor) {
this.jobId = jobId;
this.jobMastersToWaitFor = new CountDownLatch(numJobMastersToWaitFor);
@@ -430,6 +454,8 @@ public class MiniClusterJobDispatcher {
if (runnerException == null) {
runnerException = exception;
}
+
+ jobMastersToWaitFor.countDown();
}
public JobExecutionResult getResult() throws JobExecutionException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
index 9b2e318..089e4b0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
@@ -130,13 +130,13 @@ public interface RpcService {
/**
* Gets the executor, provided by this RPC service. This executor can be used for example for
* the {@code handleAsync(...)} or {@code thenAcceptAsync(...)} methods of futures.
- *
+ *
* <p><b>IMPORTANT:</b> This executor does not isolate the method invocations against
* any concurrent invocations and is therefore not suitable to run completion methods of futures
* that modify state of an {@link RpcEndpoint}. For such operations, one needs to use the
* {@link RpcEndpoint#getMainThreadExecutor() MainThreadExecutionContext} of that
* {@code RpcEndpoint}.
- *
+ *
* @return The execution context provided by the RPC service
*/
Executor getExecutor();
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index 6ea6383..46cd9e2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.dispatcher;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.BlobServerOptions;
@@ -35,7 +36,6 @@ import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
@@ -129,12 +129,12 @@ public class DispatcherTest extends TestLogger {
private TestingDispatcher dispatcher;
@BeforeClass
- public static void setup() {
+ public static void setupClass() {
rpcService = new TestingRpcService();
}
@AfterClass
- public static void teardown() {
+ public static void teardownClass() {
if (rpcService != null) {
rpcService.stopService();
@@ -274,10 +274,7 @@ public class DispatcherTest extends TestLogger {
final DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
- OnCompletionActions onCompletionActions;
-
final JobID failedJobId = new JobID();
- onCompletionActions = dispatcher.new DispatcherOnCompleteActions(failedJobId);
final JobStatus expectedState = JobStatus.FAILED;
final ArchivedExecutionGraph failedExecutionGraph = new ArchivedExecutionGraphBuilder()
@@ -286,7 +283,7 @@ public class DispatcherTest extends TestLogger {
.setFailureCause(new ErrorInfo(new RuntimeException("expected"), 1L))
.build();
- onCompletionActions.jobReachedGloballyTerminalState(failedExecutionGraph);
+ dispatcher.completeJobExecution(failedExecutionGraph);
assertThat(
dispatcherGateway.requestJobStatus(failedJobId, TIMEOUT).get(),
@@ -398,6 +395,12 @@ public class DispatcherTest extends TestLogger {
super.recoverJobs();
}
}
+
+ @VisibleForTesting
+ void completeJobExecution(ArchivedExecutionGraph archivedExecutionGraph) {
+ runAsync(
+ () -> jobReachedGloballyTerminalState(archivedExecutionGraph));
+ }
}
private static final class ExpectedJobIdJobManagerRunnerFactory implements Dispatcher.JobManagerRunnerFactory {
@@ -419,8 +422,6 @@ public class DispatcherTest extends TestLogger {
BlobServer blobServer,
JobManagerSharedServices jobManagerSharedServices,
MetricRegistry metricRegistry,
- OnCompletionActions onCompleteActions,
- FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
assertEquals(expectedJobId, jobGraph.getJobID());
@@ -434,8 +435,6 @@ public class DispatcherTest extends TestLogger {
blobServer,
jobManagerSharedServices,
metricRegistry,
- onCompleteActions,
- fatalErrorHandler,
restAddress);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
index 1040eee..dfabc61 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
@@ -31,7 +31,6 @@ import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
-import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
import org.apache.flink.runtime.jobmaster.JobResult;
@@ -40,7 +39,6 @@ import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.TestingRpcService;
@@ -67,6 +65,7 @@ import java.util.concurrent.CompletableFuture;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Tests for the {@link MiniDispatcher}.
@@ -74,18 +73,15 @@ import static org.mockito.Mockito.mock;
@Category(Flip6.class)
public class MiniDispatcherTest extends TestLogger {
- private static final JobGraph jobGraph = new JobGraph();
-
- private static final ArchivedExecutionGraph archivedExecutionGraph = new ArchivedExecutionGraphBuilder()
- .setJobID(jobGraph.getJobID())
- .setState(JobStatus.FINISHED)
- .build();
-
private static final Time timeout = Time.seconds(10L);
@ClassRule
public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+ private static JobGraph jobGraph;
+
+ private static ArchivedExecutionGraph archivedExecutionGraph;
+
private static TestingRpcService rpcService;
private static Configuration configuration;
@@ -100,6 +96,8 @@ public class MiniDispatcherTest extends TestLogger {
private CompletableFuture<JobGraph> jobGraphFuture;
+ private CompletableFuture<ArchivedExecutionGraph> resultFuture;
+
private TestingLeaderElectionService dispatcherLeaderElectionService;
private TestingHighAvailabilityServices highAvailabilityServices;
@@ -110,6 +108,13 @@ public class MiniDispatcherTest extends TestLogger {
@BeforeClass
public static void setupClass() throws IOException {
+ jobGraph = new JobGraph();
+
+ archivedExecutionGraph = new ArchivedExecutionGraphBuilder()
+ .setJobID(jobGraph.getJobID())
+ .setState(JobStatus.FINISHED)
+ .build();
+
rpcService = new TestingRpcService();
configuration = new Configuration();
@@ -127,8 +132,9 @@ public class MiniDispatcherTest extends TestLogger {
highAvailabilityServices.setDispatcherLeaderElectionService(dispatcherLeaderElectionService);
jobGraphFuture = new CompletableFuture<>();
+ resultFuture = new CompletableFuture<>();
- testingJobManagerRunnerFactory = new TestingJobManagerRunnerFactory(jobGraphFuture);
+ testingJobManagerRunnerFactory = new TestingJobManagerRunnerFactory(jobGraphFuture, resultFuture);
}
@After
@@ -180,15 +186,13 @@ public class MiniDispatcherTest extends TestLogger {
miniDispatcher.start();
try {
- final Dispatcher.DispatcherOnCompleteActions completeActions = miniDispatcher.new DispatcherOnCompleteActions(jobGraph.getJobID());
-
// wait until the Dispatcher is the leader
dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get();
// wait until we have submitted the job
jobGraphFuture.get();
- completeActions.jobReachedGloballyTerminalState(archivedExecutionGraph);
+ resultFuture.complete(archivedExecutionGraph);
// wait until we terminate
miniDispatcher.getTerminationFuture().get();
@@ -208,15 +212,13 @@ public class MiniDispatcherTest extends TestLogger {
miniDispatcher.start();
try {
- final Dispatcher.DispatcherOnCompleteActions completeActions = miniDispatcher.new DispatcherOnCompleteActions(jobGraph.getJobID());
-
// wait until the Dispatcher is the leader
dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get();
// wait until we have submitted the job
jobGraphFuture.get();
- completeActions.jobReachedGloballyTerminalState(archivedExecutionGraph);
+ resultFuture.complete(archivedExecutionGraph);
final CompletableFuture<Boolean> terminationFuture = miniDispatcher.getTerminationFuture();
@@ -262,16 +264,31 @@ public class MiniDispatcherTest extends TestLogger {
private static final class TestingJobManagerRunnerFactory implements Dispatcher.JobManagerRunnerFactory {
private final CompletableFuture<JobGraph> jobGraphFuture;
+ private final CompletableFuture<ArchivedExecutionGraph> resultFuture;
- private TestingJobManagerRunnerFactory(CompletableFuture<JobGraph> jobGraphFuture) {
+ private TestingJobManagerRunnerFactory(CompletableFuture<JobGraph> jobGraphFuture, CompletableFuture<ArchivedExecutionGraph> resultFuture) {
this.jobGraphFuture = jobGraphFuture;
+ this.resultFuture = resultFuture;
}
@Override
- public JobManagerRunner createJobManagerRunner(ResourceID resourceId, JobGraph jobGraph, Configuration configuration, RpcService rpcService, HighAvailabilityServices highAvailabilityServices, HeartbeatServices heartbeatServices, BlobServer blobServer, JobManagerSharedServices jobManagerSharedServices, MetricRegistry metricRegistry, OnCompletionActions onCompleteActions, FatalErrorHandler fatalErrorHandler, @Nullable String restAddress) throws Exception {
+ public JobManagerRunner createJobManagerRunner(
+ ResourceID resourceId,
+ JobGraph jobGraph,
+ Configuration configuration,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
+ JobManagerSharedServices jobManagerSharedServices,
+ MetricRegistry metricRegistry,
+ @Nullable String restAddress) throws Exception {
jobGraphFuture.complete(jobGraph);
- return mock(JobManagerRunner.class);
+ final JobManagerRunner mock = mock(JobManagerRunner.class);
+ when(mock.getResultFuture()).thenReturn(resultFuture);
+
+ return mock;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/075f5b69/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
index 174422f..ba72293 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
@@ -18,7 +18,204 @@
package org.apache.flink.runtime.jobmaster;
-public class JobManagerRunnerTest {
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.TestingLeaderRetrievalService;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
+import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.testtasks.NoOpInvokable;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nonnull;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link JobManagerRunner}
+ */
+public class JobManagerRunnerTest extends TestLogger {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static Configuration configuration;
+
+ private static TestingRpcService rpcService;
+
+ private static BlobServer blobServer;
+
+ private static HeartbeatServices heartbeatServices = new HeartbeatServices(1000L, 1000L);
+
+ private static JobManagerSharedServices jobManagerSharedServices;
+
+ private static MetricRegistry metricRegistry;
+
+ private static JobGraph jobGraph;
+
+ private static ArchivedExecutionGraph archivedExecutionGraph;
+
+ private TestingHighAvailabilityServices haServices;
+
+ @BeforeClass
+ public static void setupClass() throws Exception {
+ configuration = new Configuration();
+ rpcService = new TestingRpcService();
+
+ configuration.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+ blobServer = new BlobServer(
+ configuration,
+ new VoidBlobStore());
+
+ jobManagerSharedServices = JobManagerSharedServices.fromConfiguration(configuration, blobServer);
+
+ metricRegistry = NoOpMetricRegistry.INSTANCE;
+
+ final JobVertex jobVertex = new JobVertex("Test vertex");
+ jobVertex.setInvokableClass(NoOpInvokable.class);
+ jobGraph = new JobGraph(jobVertex);
+
+ archivedExecutionGraph = new ArchivedExecutionGraphBuilder()
+ .setJobID(jobGraph.getJobID())
+ .setState(JobStatus.FINISHED)
+ .build();
+ }
+
+ @Before
+ public void setup() {
+ haServices = new TestingHighAvailabilityServices();
+ haServices.setJobMasterLeaderElectionService(jobGraph.getJobID(), new TestingLeaderElectionService());
+ haServices.setResourceManagerLeaderRetriever(new TestingLeaderRetrievalService());
+ haServices.setCheckpointRecoveryFactory(new StandaloneCheckpointRecoveryFactory());
+ }
+
+ @After
+ public void tearDown() {
+
+ }
+
+ @AfterClass
+ public static void tearDownClass() throws Exception {
+ if (jobManagerSharedServices != null) {
+ jobManagerSharedServices.shutdown();
+ }
+
+ if (blobServer != null) {
+ blobServer.close();
+ }
+
+ if (rpcService != null) {
+ rpcService.stopService();
+ }
+ }
- // TODO: Test that
+ @Test
+ public void testJobCompletion() throws Exception {
+ final JobManagerRunner jobManagerRunner = createJobManagerRunner();
+
+ try {
+ jobManagerRunner.start();
+
+ final CompletableFuture<ArchivedExecutionGraph> resultFuture = jobManagerRunner.getResultFuture();
+
+ assertThat(resultFuture.isDone(), is(false));
+
+ jobManagerRunner.jobReachedGloballyTerminalState(archivedExecutionGraph);
+
+ assertThat(resultFuture.get(), is(archivedExecutionGraph));
+ } finally {
+ jobManagerRunner.shutdown();
+ }
+ }
+
+ @Test
+ public void testJobFinishedByOther() throws Exception {
+ final JobManagerRunner jobManagerRunner = createJobManagerRunner();
+
+ try {
+ jobManagerRunner.start();
+
+ final CompletableFuture<ArchivedExecutionGraph> resultFuture = jobManagerRunner.getResultFuture();
+
+ assertThat(resultFuture.isDone(), is(false));
+
+ jobManagerRunner.jobFinishedByOther();
+
+ try {
+ resultFuture.get();
+ fail("Should have failed.");
+ } catch (ExecutionException ee) {
+ assertThat(ExceptionUtils.stripExecutionException(ee), instanceOf(JobNotFinishedException.class));
+ }
+ } finally {
+ jobManagerRunner.shutdown();
+ }
+ }
+
+ @Test
+ public void testShutDown() throws Exception {
+ final JobManagerRunner jobManagerRunner = createJobManagerRunner();
+
+ try {
+ jobManagerRunner.start();
+
+ final CompletableFuture<ArchivedExecutionGraph> resultFuture = jobManagerRunner.getResultFuture();
+
+ assertThat(resultFuture.isDone(), is(false));
+
+ jobManagerRunner.shutdown();
+
+ try {
+ resultFuture.get();
+ fail("Should have failed.");
+ } catch (ExecutionException ee) {
+ assertThat(ExceptionUtils.stripExecutionException(ee), instanceOf(JobNotFinishedException.class));
+ }
+ } finally {
+ jobManagerRunner.shutdown();
+ }
+ }
+
+ @Nonnull
+ private JobManagerRunner createJobManagerRunner() throws Exception {
+ return new JobManagerRunner(
+ ResourceID.generate(),
+ jobGraph,
+ configuration,
+ rpcService,
+ haServices,
+ heartbeatServices,
+ blobServer,
+ jobManagerSharedServices,
+ metricRegistry,
+ null);
+ }
}
[6/9] flink git commit: [FLINK-8614] [flip6] Activate Flip-6 mode per
default
Posted by tr...@apache.org.
[FLINK-8614] [flip6] Activate Flip-6 mode per default
This commit enables the Flip-6 mode per default. Additionally, it disables
some of the Yarn tests which no longer apply to Flip-6 (tests which wait for
a number of started TM container without a job submission).
This closes #5437.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ab8316f3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ab8316f3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ab8316f3
Branch: refs/heads/master
Commit: ab8316f31a4de0648ab3ffa9f19fcd419aaa3bb9
Parents: ca9ee35
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Jan 30 09:22:03 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:54 2018 +0100
----------------------------------------------------------------------
.../program/rest/RestClusterClientTest.java | 2 +-
.../apache/flink/configuration/CoreOptions.java | 12 ++--
.../flink/configuration/JobManagerOptions.java | 2 +-
flink-dist/src/main/flink-bin/bin/config.sh | 2 +-
.../flink/docs/rest/RestAPIDocGenerator.java | 2 +-
.../dispatcher/DispatcherRestEndpoint.java | 3 +-
.../runtime/jobmaster/JobManagerRunner.java | 2 +-
.../flink/runtime/jobmaster/JobMaster.java | 27 ++++----
.../jobmaster/MiniDispatcherRestEndpoint.java | 3 +-
.../flink/runtime/rest/FileUploadHandler.java | 44 +------------
.../apache/flink/runtime/rest/RestClient.java | 2 +-
.../flink/runtime/rest/RestServerEndpoint.java | 42 +++++++++++-
.../runtime/webmonitor/WebMonitorEndpoint.java | 2 +-
.../runtime/rest/FileUploadHandlerTest.java | 67 --------------------
.../runtime/rest/RestServerEndpointITCase.java | 2 +-
.../runtime/rest/RestServerEndpointTest.java | 37 +++++++++++
.../flink/yarn/YARNHighAvailabilityITCase.java | 3 +
.../YARNSessionCapacitySchedulerITCase.java | 12 ++--
.../flink/yarn/YARNSessionFIFOITCase.java | 10 +--
.../org/apache/flink/yarn/YarnTestBase.java | 4 ++
.../apache/flink/yarn/YarnResourceManager.java | 2 +-
.../flink/yarn/cli/FlinkYarnSessionCli.java | 2 +-
22 files changed, 135 insertions(+), 149 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
index 9e75eea..f54d9d2 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
@@ -578,7 +578,7 @@ public class RestClusterClientTest extends TestLogger {
private final AbstractRestHandler<?, ?, ?, ?>[] abstractRestHandlers;
- TestRestServerEndpoint(final AbstractRestHandler<?, ?, ?, ?>... abstractRestHandlers) {
+ TestRestServerEndpoint(final AbstractRestHandler<?, ?, ?, ?>... abstractRestHandlers) throws IOException {
super(restServerEndpointConfiguration);
this.abstractRestHandlers = abstractRestHandlers;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
index c44219f..9bd7fab 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
@@ -230,15 +230,15 @@ public class CoreOptions {
// ------------------------------------------------------------------------
/**
+ * Constant value for the Flip-6 execution mode.
+ */
+ public static final String FLIP6_MODE = "flip6";
+
+ /**
* Switch to select the execution mode. Possible values are 'flip6' and 'old'.
*/
public static final ConfigOption<String> MODE = ConfigOptions
.key("mode")
- .defaultValue("old")
+ .defaultValue(FLIP6_MODE)
.withDescription("Switch to select the execution mode. Possible values are 'flip6' and 'old'.");
-
- /**
- * Constant value for the Flip-6 execution mode.
- */
- public static final String FLIP6_MODE = "flip6";
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index 74ec5b7..ade3958 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -140,7 +140,7 @@ public class JobManagerOptions {
public static final ConfigOption<Long> SLOT_IDLE_TIMEOUT =
key("slot.idle.timeout")
- .defaultValue(20L * 1000L)
+ .defaultValue(10L * 1000L)
.withDescription("The timeout in milliseconds for a idle slot in Slot Pool.");
// ---------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-dist/src/main/flink-bin/bin/config.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 807bf33..645e156 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -267,7 +267,7 @@ fi
# Define FLIP if it is not already set
if [ -z "${FLINK_MODE}" ]; then
- FLINK_MODE=$(readFromConfig ${KEY_FLINK_MODE} "old" "${YAML_CONF}")
+ FLINK_MODE=$(readFromConfig ${KEY_FLINK_MODE} "flip6" "${YAML_CONF}")
fi
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
----------------------------------------------------------------------
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java b/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
index f4e1c70..b3a5def 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
@@ -288,7 +288,7 @@ public class RestAPIDocGenerator {
metricQueryServiceRetriever = path -> null;
}
- private DocumentingDispatcherRestEndpoint() {
+ private DocumentingDispatcherRestEndpoint() throws IOException {
super(
restConfig,
dispatcherGatewayRetriever,
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
index b84e287..f8245c2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
@@ -38,6 +38,7 @@ import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
@@ -61,7 +62,7 @@ public class DispatcherRestEndpoint extends WebMonitorEndpoint<DispatcherGateway
Executor executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
- FatalErrorHandler fatalErrorHandler) {
+ FatalErrorHandler fatalErrorHandler) throws IOException {
super(
endpointConfiguration,
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
index 285cb4a..5740bd7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
@@ -388,7 +388,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
log.info("JobManager for job {} ({}) was revoked leadership at {}.",
jobGraph.getName(), jobGraph.getJobID(), getAddress());
- CompletableFuture<Acknowledge> suspendFuture = jobManager.suspend(new Exception("JobManager is no longer the leader."), rpcTimeout);
+ CompletableFuture<Acknowledge> suspendFuture = jobManager.suspend(new FlinkException("JobManager is no longer the leader."), rpcTimeout);
suspendFuture.whenCompleteAsync(
(Acknowledge ack, Throwable throwable) -> {
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index aeac2df..139c053 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -347,7 +347,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
* @param timeout for this operation
* @return Future acknowledge indicating that the job has been suspended. Otherwise the future contains an exception
*/
- public CompletableFuture<Acknowledge> suspend(final Throwable cause, final Time timeout) {
+ public CompletableFuture<Acknowledge> suspend(final Exception cause, final Time timeout) {
CompletableFuture<Acknowledge> suspendFuture = callAsyncWithoutFencing(() -> suspendExecution(cause), timeout);
stop();
@@ -375,7 +375,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
resourceManagerHeartbeatManager.stop();
// make sure there is a graceful exit
- suspendExecution(new Exception("JobManager is shutting down."));
+ suspendExecution(new FlinkException("JobManager is shutting down."));
// shut down will internally release all registered slots
slotPool.shutDown();
@@ -595,14 +595,11 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
if (checkpointCoordinator != null) {
- getRpcService().execute(new Runnable() {
- @Override
- public void run() {
- try {
- checkpointCoordinator.receiveDeclineMessage(decline);
- } catch (Exception e) {
- log.error("Error in CheckpointCoordinator while processing {}", decline, e);
- }
+ getRpcService().execute(() -> {
+ try {
+ checkpointCoordinator.receiveDeclineMessage(decline);
+ } catch (Exception e) {
+ log.error("Error in CheckpointCoordinator while processing {}", decline, e);
}
});
} else {
@@ -915,7 +912,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
*
* @param cause The reason of why this job been suspended.
*/
- private Acknowledge suspendExecution(final Throwable cause) {
+ private Acknowledge suspendExecution(final Exception cause) {
validateRunsInMainThread();
if (getFencingToken() == null) {
@@ -939,7 +936,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
slotPoolGateway.suspend();
// disconnect from resource manager:
- closeResourceManagerConnection(new Exception("Execution was suspended.", cause));
+ closeResourceManagerConnection(cause);
return Acknowledge.get();
}
@@ -1037,7 +1034,11 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private void closeResourceManagerConnection(Exception cause) {
if (resourceManagerConnection != null) {
- log.info("Close ResourceManager connection {}.", resourceManagerConnection.getResourceManagerResourceID(), cause);
+ if (log.isDebugEnabled()) {
+ log.debug("Close ResourceManager connection {}.", resourceManagerConnection.getResourceManagerResourceID(), cause);
+ } else {
+ log.info("Close ResourceManager connection {}: {}.", resourceManagerConnection.getResourceManagerResourceID(), cause.getMessage());
+ }
resourceManagerHeartbeatManager.unmonitorTarget(resourceManagerConnection.getResourceManagerResourceID());
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
index 54ad526..b31c46a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
+import java.io.IOException;
import java.util.concurrent.Executor;
/**
@@ -48,7 +49,7 @@ public class MiniDispatcherRestEndpoint extends WebMonitorEndpoint<RestfulGatewa
Executor executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
- FatalErrorHandler fatalErrorHandler) {
+ FatalErrorHandler fatalErrorHandler) throws IOException {
super(
endpointConfiguration,
leaderRetriever,
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
index c214b50..8854a1f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.rest;
-import org.apache.flink.annotation.VisibleForTesting;
-
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
@@ -39,8 +37,6 @@ import org.apache.flink.shaded.netty4.io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.UUID;
@@ -66,10 +62,8 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
private HttpRequest currentHttpRequest;
- public FileUploadHandler(final Path uploadDir) throws IOException {
+ public FileUploadHandler(final Path uploadDir) {
super(false);
- createUploadDir(uploadDir);
-
DiskFileUpload.baseDirectory = uploadDir.normalize().toAbsolutePath().toString();
this.uploadDir = requireNonNull(uploadDir);
}
@@ -89,7 +83,8 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
ctx.fireChannelRead(msg);
}
} else if (msg instanceof HttpContent && currentHttpPostRequestDecoder != null) {
- createUploadDir(uploadDir);
+ // make sure that we still have a upload dir in case that it got deleted in the meanwhile
+ RestServerEndpoint.createUploadDir(uploadDir, LOG);
final HttpContent httpContent = (HttpContent) msg;
currentHttpPostRequestDecoder.offer(httpContent);
@@ -123,37 +118,4 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
currentHttpPostRequestDecoder = null;
currentHttpRequest = null;
}
-
- /**
- * Creates the upload dir if needed.
- */
- @VisibleForTesting
- static void createUploadDir(final Path uploadDir) throws IOException {
- if (!Files.exists(uploadDir)) {
- LOG.warn("Upload directory {} does not exist, or has been deleted externally. " +
- "Previously uploaded files are no longer available.", uploadDir);
- checkAndCreateUploadDir(uploadDir);
- }
- }
-
- /**
- * Checks whether the given directory exists and is writable. If it doesn't exist, this method
- * will attempt to create it.
- *
- * @param uploadDir directory to check
- * @throws IOException if the directory does not exist and cannot be created, or if the
- * directory isn't writable
- */
- private static synchronized void checkAndCreateUploadDir(final Path uploadDir) throws IOException {
- if (Files.exists(uploadDir) && Files.isWritable(uploadDir)) {
- LOG.info("Using directory {} for file uploads.", uploadDir);
- } else if (Files.isWritable(Files.createDirectories(uploadDir))) {
- LOG.info("Created directory {} for file uploads.", uploadDir);
- } else {
- LOG.warn("Upload directory {} cannot be created or is not writable.", uploadDir);
- throw new IOException(
- String.format("Upload directory %s cannot be created or is not writable.",
- uploadDir));
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
index 43a9981..42612ea 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
@@ -151,7 +151,7 @@ public class RestClient {
String targetUrl = MessageParameters.resolveUrl(messageHeaders.getTargetRestEndpointURL(), messageParameters);
- LOG.debug("Sending request of class {} to {}", request.getClass(), targetUrl);
+ LOG.debug("Sending request of class {} to {}:{}{}", request.getClass(), targetAddress, targetPort, targetUrl);
// serialize payload
StringWriter sw = new StringWriter();
objectMapper.writeValue(sw, request);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index 1904235..dbb25a7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.rest;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.rest.handler.PipelineErrorHandler;
@@ -49,6 +50,7 @@ import javax.net.ssl.SSLEngine;
import java.io.IOException;
import java.io.Serializable;
import java.net.InetSocketAddress;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
import java.util.Comparator;
@@ -77,12 +79,14 @@ public abstract class RestServerEndpoint {
private volatile boolean started;
- public RestServerEndpoint(RestServerEndpointConfiguration configuration) {
+ public RestServerEndpoint(RestServerEndpointConfiguration configuration) throws IOException {
Preconditions.checkNotNull(configuration);
this.configuredAddress = configuration.getEndpointBindAddress();
this.configuredPort = configuration.getEndpointBindPort();
this.sslEngine = configuration.getSslEngine();
+
this.uploadDir = configuration.getUploadDir();
+ createUploadDir(uploadDir, log);
this.restAddress = null;
@@ -136,7 +140,7 @@ public abstract class RestServerEndpoint {
ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
@Override
- protected void initChannel(SocketChannel ch) throws IOException {
+ protected void initChannel(SocketChannel ch) {
Handler handler = new RouterHandler(router);
// SSL should be the first handler in the pipeline
@@ -318,6 +322,40 @@ public abstract class RestServerEndpoint {
}
/**
+ * Creates the upload dir if needed.
+ */
+ @VisibleForTesting
+ static void createUploadDir(final Path uploadDir, final Logger log) throws IOException {
+ if (!Files.exists(uploadDir)) {
+ log.warn("Upload directory {} does not exist, or has been deleted externally. " +
+ "Previously uploaded files are no longer available.", uploadDir);
+ checkAndCreateUploadDir(uploadDir, log);
+ }
+ }
+
+ /**
+ * Checks whether the given directory exists and is writable. If it doesn't exist, this method
+ * will attempt to create it.
+ *
+ * @param uploadDir directory to check
+ * @param log logger used for logging output
+ * @throws IOException if the directory does not exist and cannot be created, or if the
+ * directory isn't writable
+ */
+ private static synchronized void checkAndCreateUploadDir(final Path uploadDir, final Logger log) throws IOException {
+ if (Files.exists(uploadDir) && Files.isWritable(uploadDir)) {
+ log.info("Using directory {} for file uploads.", uploadDir);
+ } else if (Files.isWritable(Files.createDirectories(uploadDir))) {
+ log.info("Created directory {} for file uploads.", uploadDir);
+ } else {
+ log.warn("Upload directory {} cannot be created or is not writable.", uploadDir);
+ throw new IOException(
+ String.format("Upload directory %s cannot be created or is not writable.",
+ uploadDir));
+ }
+ }
+
+ /**
* Comparator for Rest URLs.
*
* <p>The comparator orders the Rest URLs such that URLs with path parameters are ordered behind
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index 5e4c72b..10ad344 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -163,7 +163,7 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
Executor executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
- FatalErrorHandler fatalErrorHandler) {
+ FatalErrorHandler fatalErrorHandler) throws IOException {
super(endpointConfiguration);
this.leaderRetriever = Preconditions.checkNotNull(leaderRetriever);
this.clusterConfiguration = Preconditions.checkNotNull(clusterConfiguration);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/test/java/org/apache/flink/runtime/rest/FileUploadHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/FileUploadHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/FileUploadHandlerTest.java
deleted file mode 100644
index 61277ae..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/FileUploadHandlerTest.java
+++ /dev/null
@@ -1,67 +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.rest;
-
-import org.apache.flink.util.TestLogger;
-
-import org.junit.Assume;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/**
- * Tests for {@link FileUploadHandler}.
- */
-public class FileUploadHandlerTest extends TestLogger {
-
- @Rule
- public TemporaryFolder temporaryFolder = new TemporaryFolder();
-
- @Test
- public void testCreateUploadDir() throws Exception {
- final File file = temporaryFolder.newFolder();
- final Path testUploadDir = file.toPath().resolve("testUploadDir");
- assertFalse(Files.exists(testUploadDir));
- FileUploadHandler.createUploadDir(testUploadDir);
- assertTrue(Files.exists(testUploadDir));
- }
-
- @Test
- public void testCreateUploadDirFails() throws Exception {
- final File file = temporaryFolder.newFolder();
- Assume.assumeTrue(file.setWritable(false));
-
- final Path testUploadDir = file.toPath().resolve("testUploadDir");
- assertFalse(Files.exists(testUploadDir));
- try {
- FileUploadHandler.createUploadDir(testUploadDir);
- fail("Expected exception not thrown.");
- } catch (IOException e) {
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
index 735b12d..50b26e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
@@ -303,7 +303,7 @@ public class RestServerEndpointITCase extends TestLogger {
TestRestServerEndpoint(
RestServerEndpointConfiguration configuration,
TestHandler testHandler,
- TestUploadHandler testUploadHandler) {
+ TestUploadHandler testUploadHandler) throws IOException {
super(configuration);
this.testHandler = Preconditions.checkNotNull(testHandler);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointTest.java
index 866dfbf..de32883 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointTest.java
@@ -21,14 +21,25 @@ package org.apache.flink.runtime.rest;
import org.apache.flink.testutils.category.Flip6;
import org.apache.flink.util.TestLogger;
+import org.junit.Assume;
+import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.helpers.NOPLogger;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/**
* Test cases for the {@link RestServerEndpoint}.
@@ -36,6 +47,9 @@ import static org.junit.Assert.assertEquals;
@Category(Flip6.class)
public class RestServerEndpointTest extends TestLogger {
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
/**
* Tests that the REST handler URLs are properly sorted.
*/
@@ -63,4 +77,27 @@ public class RestServerEndpointTest extends TestLogger {
assertEquals(expected, handlerUrls);
}
+
+ @Test
+ public void testCreateUploadDir() throws Exception {
+ final File file = temporaryFolder.newFolder();
+ final Path testUploadDir = file.toPath().resolve("testUploadDir");
+ assertFalse(Files.exists(testUploadDir));
+ RestServerEndpoint.createUploadDir(testUploadDir, NOPLogger.NOP_LOGGER);
+ assertTrue(Files.exists(testUploadDir));
+ }
+
+ @Test
+ public void testCreateUploadDirFails() throws Exception {
+ final File file = temporaryFolder.newFolder();
+ Assume.assumeTrue(file.setWritable(false));
+
+ final Path testUploadDir = file.toPath().resolve("testUploadDir");
+ assertFalse(Files.exists(testUploadDir));
+ try {
+ RestServerEndpoint.createUploadDir(testUploadDir, NOPLogger.NOP_LOGGER);
+ fail("Expected exception not thrown.");
+ } catch (IOException e) {
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index 5a5fb74..05be03a 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -55,6 +55,8 @@ import java.util.concurrent.TimeUnit;
import scala.concurrent.duration.FiniteDuration;
+import static org.junit.Assume.assumeTrue;
+
/**
* Tests that verify correct HA behavior.
*/
@@ -104,6 +106,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
*/
@Test
public void testMultipleAMKill() throws Exception {
+ assumeTrue("This test only works with the old actor based code.", !flip6);
final int numberKillingAttempts = numberApplicationAttempts - 1;
String confDirPath = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
final Configuration configuration = GlobalConfiguration.loadConfiguration();
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index c806c5e..3a674ad 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -78,6 +78,7 @@ import java.util.regex.Pattern;
import static junit.framework.TestCase.assertTrue;
import static org.apache.flink.yarn.UtilsTest.addTestAppender;
import static org.apache.flink.yarn.UtilsTest.checkForLogString;
+import static org.junit.Assume.assumeTrue;
/**
* This test starts a MiniYARNCluster with a CapacityScheduler.
@@ -101,6 +102,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
*/
@Test
public void testClientStartup() throws IOException {
+ assumeTrue("Flip-6 does not start TMs upfront.", !flip6);
LOG.info("Starting testClientStartup()");
runWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(), "-t", flinkLibFolder.getAbsolutePath(),
"-n", "1",
@@ -130,7 +132,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
"-yjm", "768",
"-ytm", "1024", exampleJarLocation.getAbsolutePath()},
/* test succeeded after this string */
- "Job execution complete",
+ "Program execution finished",
/* prohibited strings: (to verify the parallelism) */
// (we should see "DataSink (...) (1/2)" and "DataSink (...) (2/2)" instead)
new String[]{"DataSink \\(.*\\) \\(1/1\\) switched to FINISHED"},
@@ -177,7 +179,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
"-yD", "taskmanager.memory.size=" + offHeapMemory,
"-yD", "taskmanager.memory.preallocate=true", exampleJarLocation.getAbsolutePath()},
/* test succeeded after this string */
- "Job execution complete",
+ "Program execution finished",
/* prohibited strings: (to verify the parallelism) */
// (we should see "DataSink (...) (1/2)" and "DataSink (...) (2/2)" instead)
new String[]{"DataSink \\(.*\\) \\(1/1\\) switched to FINISHED"},
@@ -190,6 +192,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
*/
@Test(timeout = 100000) // timeout after 100 seconds
public void testTaskManagerFailure() throws Exception {
+ assumeTrue("Flip-6 does not start TMs upfront.", !flip6);
LOG.info("Starting testTaskManagerFailure()");
Runner runner = startWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(), "-t", flinkLibFolder.getAbsolutePath(),
"-n", "1",
@@ -402,7 +405,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
"-yjm", "768",
"-ytm", "1024", exampleJarLocation.getAbsolutePath()},
/* test succeeded after this string */
- "Job execution complete",
+ "Program execution finished",
/* prohibited strings: (we want to see "DataSink (...) (2/2) switched to FINISHED") */
new String[]{"DataSink \\(.*\\) \\(1/1\\) switched to FINISHED"},
RunTypes.CLI_FRONTEND, 0, true);
@@ -477,7 +480,8 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
"-yD", "yarn.tags=test-tag",
"-ytm", "1024",
"-ys", "2", // test requesting slots from YARN.
- "--yarndetached", job,
+ "-p", "2",
+ "--detached", job,
"--input", tmpInFile.getAbsoluteFile().toString(),
"--output", tmpOutFolder.getAbsoluteFile().toString()},
"Job has been submitted with JobID",
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index 975dd28..e545187 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -98,10 +98,12 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
runner.join();
checkForLogString("The Flink YARN client has been started in detached mode");
- LOG.info("Waiting until two containers are running");
- // wait until two containers are running
- while (getRunningContainers() < 2) {
- sleep(500);
+ if (!flip6) {
+ LOG.info("Waiting until two containers are running");
+ // wait until two containers are running
+ while (getRunningContainers() < 2) {
+ sleep(500);
+ }
}
//additional sleep for the JM/TM to start and establish connection
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
index eeda32d..c863e14 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
@@ -20,6 +20,7 @@ package org.apache.flink.yarn;
import org.apache.flink.client.cli.CliFrontend;
import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.SecurityOptions;
import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.util.Preconditions;
@@ -187,6 +188,7 @@ public abstract class YarnTestBase extends TestLogger {
private YarnClient yarnClient = null;
protected org.apache.flink.configuration.Configuration flinkConfiguration;
+ protected boolean flip6;
@Before
public void checkClusterEmpty() throws IOException, YarnException {
@@ -207,6 +209,8 @@ public abstract class YarnTestBase extends TestLogger {
}
flinkConfiguration = new org.apache.flink.configuration.Configuration();
+
+ flip6 = CoreOptions.FLIP6_MODE.equalsIgnoreCase(flinkConfiguration.getString(CoreOptions.MODE));
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index d07ac5a..87324cb 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -432,7 +432,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
final ContaineredTaskManagerParameters taskManagerParameters =
ContaineredTaskManagerParameters.create(flinkConfig, resource.getMemory(), 1);
- log.info("TaskExecutor{} will be started with container size {} MB, JVM heap size {} MB, " +
+ log.info("TaskExecutor {} will be started with container size {} MB, JVM heap size {} MB, " +
"JVM direct memory limit {} MB",
containerId,
taskManagerParameters.taskManagerTotalMemoryMB(),
http://git-wip-us.apache.org/repos/asf/flink/blob/ab8316f3/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index a289b66..8882236 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -365,7 +365,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId
}
private ClusterSpecification createClusterSpecification(Configuration configuration, CommandLine cmd) {
- if (!cmd.hasOption(container.getOpt())) { // number of containers is required option!
+ if (!flip6 && !cmd.hasOption(container.getOpt())) { // number of containers is required option!
LOG.error("Missing required argument {}", container.getOpt());
printUsage();
throw new IllegalArgumentException("Missing required argument " + container.getOpt());
[9/9] flink git commit: [FLINK-8546] [flip6] Respect savepoints and
restore from latest checkpoints
Posted by tr...@apache.org.
[FLINK-8546] [flip6] Respect savepoints and restore from latest checkpoints
Let the JobMaster respect checkpoints and savepoints. The JobMaster will always
try to restore the latest checkpoint if there is one available. Next it will check
whether savepoint restore settings have been set. If so, then it will try to restore
the savepoint. Only if these settings are not set, the job will be started from
scratch.
This closes #5444.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3a4e8964
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3a4e8964
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3a4e8964
Branch: refs/heads/master
Commit: 3a4e89643d7d7642dde9b5644491f261d4d545bd
Parents: 9e85bb0
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 13 15:19:18 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:55 2018 +0100
----------------------------------------------------------------------
.../checkpoint/CheckpointCoordinator.java | 4 +-
.../flink/runtime/jobmaster/JobMaster.java | 23 ++
.../TestingCheckpointRecoveryFactory.java | 46 ++++
.../jobmanager/JobManagerHARecoveryTest.java | 24 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 217 ++++++++++++++++---
5 files changed, 260 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3a4e8964/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index ed3570a..016defb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -966,7 +966,7 @@ public class CheckpointCoordinator {
* Restores the latest checkpointed state.
*
* @param tasks Map of job vertices to restore. State for these vertices is
- * restored via {@link Execution#setInitialState(TaskStateSnapshot)}.
+ * restored via {@link Execution#setInitialState(JobManagerTaskRestore)}.
* @param errorIfNoCheckpoint Fail if no completed checkpoint is available to
* restore from.
* @param allowNonRestoredState Allow checkpoint state that cannot be mapped
@@ -1065,7 +1065,7 @@ public class CheckpointCoordinator {
* mapped to any job vertex in tasks.
* @param tasks Map of job vertices to restore. State for these
* vertices is restored via
- * {@link Execution#setInitialState(TaskStateSnapshot)}.
+ * {@link Execution#setInitialState(JobManagerTaskRestore)}.
* @param userClassLoader The class loader to resolve serialized classes in
* legacy savepoint versions.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/3a4e8964/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index dfa4d1c..2a4b881 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -57,6 +57,7 @@ import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException;
import org.apache.flink.runtime.jobmaster.message.ClassloadingProps;
@@ -297,6 +298,28 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
jobMasterConfiguration.getSlotRequestTimeout(),
log);
+ final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
+
+ if (checkpointCoordinator != null) {
+ // check whether we find a valid checkpoint
+ if (!checkpointCoordinator.restoreLatestCheckpointedState(
+ executionGraph.getAllVertices(),
+ false,
+ false)) {
+
+ // check whether we can restore from a savepoint
+ final SavepointRestoreSettings savepointRestoreSettings = jobGraph.getSavepointRestoreSettings();
+
+ if (savepointRestoreSettings.restoreSavepoint()) {
+ checkpointCoordinator.restoreSavepoint(
+ savepointRestoreSettings.getRestorePath(),
+ savepointRestoreSettings.allowNonRestoredState(),
+ executionGraph.getAllVertices(),
+ userCodeLoader);
+ }
+ }
+ }
+
// register self as job status change listener
executionGraph.registerJobStatusListener(new JobManagerJobStatusListener());
http://git-wip-us.apache.org/repos/asf/flink/blob/3a4e8964/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointRecoveryFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointRecoveryFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointRecoveryFactory.java
new file mode 100644
index 0000000..7bc0c85
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointRecoveryFactory.java
@@ -0,0 +1,46 @@
+/*
+ * 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.checkpoint;
+
+import org.apache.flink.api.common.JobID;
+
+/**
+ * Simple {@link CheckpointRecoveryFactory} which is initialized with a
+ * {@link CompletedCheckpointStore} and a {@link CheckpointIDCounter}.
+ */
+public class TestingCheckpointRecoveryFactory implements CheckpointRecoveryFactory {
+
+ private final CompletedCheckpointStore store;
+ private final CheckpointIDCounter counter;
+
+ public TestingCheckpointRecoveryFactory(CompletedCheckpointStore store, CheckpointIDCounter counter) {
+ this.store = store;
+ this.counter = counter;
+ }
+
+ @Override
+ public CompletedCheckpointStore createCheckpointStore(JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader) throws Exception {
+ return store;
+ }
+
+ @Override
+ public CheckpointIDCounter createCheckpointIDCounter(JobID jobId) throws Exception {
+ return counter;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3a4e8964/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
index 93cde3a..8f1e12c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerHARecoveryTest.java
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointIDCounter;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.checkpoint.TestingCheckpointRecoveryFactory;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
@@ -177,7 +178,7 @@ public class JobManagerHARecoveryTest extends TestLogger {
submittedJobGraphStore.start(null);
CompletedCheckpointStore checkpointStore = new RecoverableCompletedCheckpointStore();
CheckpointIDCounter checkpointCounter = new StandaloneCheckpointIDCounter();
- CheckpointRecoveryFactory checkpointStateFactory = new MyCheckpointRecoveryFactory(checkpointStore, checkpointCounter);
+ CheckpointRecoveryFactory checkpointStateFactory = new TestingCheckpointRecoveryFactory(checkpointStore, checkpointCounter);
TestingLeaderElectionService myLeaderElectionService = new TestingLeaderElectionService();
TestingLeaderRetrievalService myLeaderRetrievalService = new TestingLeaderRetrievalService(
null,
@@ -465,27 +466,6 @@ public class JobManagerHARecoveryTest extends TestLogger {
}
}
- static class MyCheckpointRecoveryFactory implements CheckpointRecoveryFactory {
-
- private final CompletedCheckpointStore store;
- private final CheckpointIDCounter counter;
-
- public MyCheckpointRecoveryFactory(CompletedCheckpointStore store, CheckpointIDCounter counter) {
- this.store = store;
- this.counter = counter;
- }
-
- @Override
- public CompletedCheckpointStore createCheckpointStore(JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader) throws Exception {
- return store;
- }
-
- @Override
- public CheckpointIDCounter createCheckpointIDCounter(JobID jobId) throws Exception {
- return counter;
- }
- }
-
public static class BlockingInvokable extends AbstractInvokable {
private static final OneShotLatch LATCH = new OneShotLatch();
http://git-wip-us.apache.org/repos/asf/flink/blob/3a4e8964/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index e142d9c..e401020 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -25,13 +25,25 @@ import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.checkpoint.CheckpointProperties;
+import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy;
+import org.apache.flink.runtime.checkpoint.Checkpoints;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointIDCounter;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.TestingCheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointV2;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.leaderelection.TestingLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
@@ -40,6 +52,8 @@ import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
+import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
@@ -57,7 +71,12 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
+import javax.annotation.Nonnull;
+
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
+import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -162,21 +181,7 @@ public class JobMasterTest extends TestLogger {
final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build();
final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
- final JobMaster jobMaster = new JobMaster(
- rpcService,
- jobMasterConfiguration,
- jmResourceId,
- jobGraph,
- haServices,
- jobManagerSharedServices,
- fastHeartbeatServices,
- blobServer,
- null,
- new NoOpOnCompletionActions(),
- testingFatalErrorHandler,
- JobMasterTest.class.getClassLoader(),
- null,
- null);
+ final JobMaster jobMaster = createJobMaster(jobMasterConfiguration, jobGraph, haServices, jobManagerSharedServices);
CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
@@ -237,21 +242,7 @@ public class JobMasterTest extends TestLogger {
final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build();
final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
- final JobMaster jobMaster = new JobMaster(
- rpcService,
- jobMasterConfiguration,
- jmResourceId,
- jobGraph,
- haServices,
- jobManagerSharedServices,
- fastHeartbeatServices,
- blobServer,
- null,
- new NoOpOnCompletionActions(),
- testingFatalErrorHandler,
- JobMasterTest.class.getClassLoader(),
- null,
- null);
+ final JobMaster jobMaster = createJobMaster(jobMasterConfiguration, jobGraph, haServices, jobManagerSharedServices);
CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
@@ -282,6 +273,152 @@ public class JobMasterTest extends TestLogger {
}
/**
+ * Tests that a JobMaster will restore the given JobGraph from its savepoint upon
+ * initial submission.
+ */
+ @Test
+ public void testRestoringFromSavepoint() throws Exception {
+
+ // create savepoint data
+ final long savepointId = 42L;
+ final File savepointFile = createSavepoint(savepointId);
+
+ // set savepoint settings
+ final SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.forPath(
+ savepointFile.getAbsolutePath(),
+ true);
+ final JobGraph jobGraph = createJobGraphWithCheckpointing(savepointRestoreSettings);
+
+ final StandaloneCompletedCheckpointStore completedCheckpointStore = new StandaloneCompletedCheckpointStore(1);
+ final TestingCheckpointRecoveryFactory testingCheckpointRecoveryFactory = new TestingCheckpointRecoveryFactory(completedCheckpointStore, new StandaloneCheckpointIDCounter());
+ haServices.setCheckpointRecoveryFactory(testingCheckpointRecoveryFactory);
+ final JobMaster jobMaster = createJobMaster(
+ JobMasterConfiguration.fromConfiguration(configuration),
+ jobGraph,
+ haServices,
+ new TestingJobManagerSharedServicesBuilder().build());
+
+ try {
+ // starting the JobMaster should have read the savepoint
+ final CompletedCheckpoint savepointCheckpoint = completedCheckpointStore.getLatestCheckpoint();
+
+ assertThat(savepointCheckpoint, Matchers.notNullValue());
+
+ assertThat(savepointCheckpoint.getCheckpointID(), Matchers.is(savepointId));
+ } finally {
+ RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
+ }
+ }
+
+ /**
+ * Tests that an existing checkpoint will have precedence over an savepoint
+ */
+ @Test
+ public void testCheckpointPrecedesSavepointRecovery() throws Exception {
+
+ // create savepoint data
+ final long savepointId = 42L;
+ final File savepointFile = createSavepoint(savepointId);
+
+ // set savepoint settings
+ final SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.forPath("" +
+ savepointFile.getAbsolutePath(),
+ true);
+ final JobGraph jobGraph = createJobGraphWithCheckpointing(savepointRestoreSettings);
+
+ final long checkpointId = 1L;
+
+ final CompletedCheckpoint completedCheckpoint = new CompletedCheckpoint(
+ jobGraph.getJobID(),
+ checkpointId,
+ 1L,
+ 1L,
+ Collections.emptyMap(),
+ null,
+ CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
+ new DummyCheckpointStorageLocation());
+
+ final StandaloneCompletedCheckpointStore completedCheckpointStore = new StandaloneCompletedCheckpointStore(1);
+ completedCheckpointStore.addCheckpoint(completedCheckpoint);
+ final TestingCheckpointRecoveryFactory testingCheckpointRecoveryFactory = new TestingCheckpointRecoveryFactory(completedCheckpointStore, new StandaloneCheckpointIDCounter());
+ haServices.setCheckpointRecoveryFactory(testingCheckpointRecoveryFactory);
+ final JobMaster jobMaster = createJobMaster(
+ JobMasterConfiguration.fromConfiguration(configuration),
+ jobGraph,
+ haServices,
+ new TestingJobManagerSharedServicesBuilder().build());
+
+ try {
+ // starting the JobMaster should have read the savepoint
+ final CompletedCheckpoint savepointCheckpoint = completedCheckpointStore.getLatestCheckpoint();
+
+ assertThat(savepointCheckpoint, Matchers.notNullValue());
+
+ assertThat(savepointCheckpoint.getCheckpointID(), Matchers.is(checkpointId));
+ } finally {
+ RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
+ }
+ }
+
+ private File createSavepoint(long savepointId) throws IOException {
+ final File savepointFile = temporaryFolder.newFile();
+ final SavepointV2 savepoint = new SavepointV2(savepointId, Collections.emptyList(), Collections.emptyList());
+
+ try (FileOutputStream fileOutputStream = new FileOutputStream(savepointFile)) {
+ Checkpoints.storeCheckpointMetadata(savepoint, fileOutputStream);
+ }
+
+ return savepointFile;
+ }
+
+ @Nonnull
+ private JobGraph createJobGraphWithCheckpointing(SavepointRestoreSettings savepointRestoreSettings) {
+ final JobGraph jobGraph = new JobGraph();
+
+ // enable checkpointing which is required to resume from a savepoint
+ final CheckpointCoordinatorConfiguration checkpoinCoordinatorConfiguration = new CheckpointCoordinatorConfiguration(
+ 1000L,
+ 1000L,
+ 1000L,
+ 1,
+ CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION,
+ true);
+ final JobCheckpointingSettings checkpointingSettings = new JobCheckpointingSettings(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ checkpoinCoordinatorConfiguration,
+ null);
+ jobGraph.setSnapshotSettings(checkpointingSettings);
+ jobGraph.setSavepointRestoreSettings(savepointRestoreSettings);
+
+ return jobGraph;
+ }
+
+ @Nonnull
+ private JobMaster createJobMaster(
+ JobMasterConfiguration jobMasterConfiguration,
+ JobGraph jobGraph,
+ HighAvailabilityServices highAvailabilityServices,
+ JobManagerSharedServices jobManagerSharedServices) throws Exception {
+ return new JobMaster(
+ rpcService,
+ jobMasterConfiguration,
+ jmResourceId,
+ jobGraph,
+ highAvailabilityServices,
+ jobManagerSharedServices,
+ fastHeartbeatServices,
+ blobServer,
+ null,
+ new NoOpOnCompletionActions(),
+ testingFatalErrorHandler,
+ JobMasterTest.class.getClassLoader(),
+ null,
+ null);
+ }
+
+ /**
* No op implementation of {@link OnCompletionActions}.
*/
private static final class NoOpOnCompletionActions implements OnCompletionActions {
@@ -297,4 +434,24 @@ public class JobMasterTest extends TestLogger {
}
}
+ private static final class DummyCheckpointStorageLocation implements CompletedCheckpointStorageLocation {
+
+ private static final long serialVersionUID = 164095949572620688L;
+
+ @Override
+ public String getExternalPointer() {
+ return null;
+ }
+
+ @Override
+ public StreamStateHandle getMetadataHandle() {
+ return null;
+ }
+
+ @Override
+ public void disposeStorageLocation() throws IOException {
+
+ }
+ }
+
}
[4/9] flink git commit: [hotfix] Only log retrying exception on debug
in RetryingRegistration
Posted by tr...@apache.org.
[hotfix] Only log retrying exception on debug in RetryingRegistration
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ca9ee35c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ca9ee35c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ca9ee35c
Branch: refs/heads/master
Commit: ca9ee35c12e902c58d5d5d8077da258663386487
Parents: 24c2429
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Feb 8 10:27:27 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:53 2018 +0100
----------------------------------------------------------------------
.../runtime/registration/RetryingRegistration.java | 17 ++++++++++++++++-
1 file changed, 16 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ca9ee35c/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
index 279714b..6959729 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
@@ -207,7 +207,22 @@ public abstract class RetryingRegistration<F extends Serializable, G extends Rpc
resourceManagerAcceptFuture.whenCompleteAsync(
(Void v, Throwable failure) -> {
if (failure != null && !canceled) {
- log.warn("Could not resolve {} address {}, retrying in {} ms", targetName, targetAddress, delayOnError, failure);
+ final Throwable strippedFailure = ExceptionUtils.stripCompletionException(failure);
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "Could not resolve {} address {}, retrying in {} ms.",
+ targetName,
+ targetAddress,
+ delayOnError,
+ strippedFailure);
+ } else {
+ log.info(
+ "Could not resolve {} address {}, retrying in {} ms: {}.",
+ targetName,
+ targetAddress,
+ delayOnError,
+ strippedFailure.getMessage());
+ }
startRegistrationLater(delayOnError);
}
[8/9] flink git commit: [hotfix] [tests] Simplify JobMasterTest
Posted by tr...@apache.org.
[hotfix] [tests] Simplify JobMasterTest
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9e85bb07
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9e85bb07
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9e85bb07
Branch: refs/heads/master
Commit: 9e85bb0780978e08f00da9c7bc4529b4dd2abafd
Parents: d8a8866
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 13 15:16:27 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:55 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/jobmaster/JobMasterTest.java | 227 ++++++++++---------
1 file changed, 126 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9e85bb07/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index e8f0bc4..e142d9c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -21,13 +21,12 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.concurrent.ScheduledExecutor;
-import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
@@ -39,6 +38,7 @@ import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
+import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
@@ -48,10 +48,16 @@ import org.apache.flink.testutils.category.Flip6;
import org.apache.flink.util.TestLogger;
import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
-import java.net.URL;
+import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -63,23 +69,85 @@ import static org.hamcrest.MatcherAssert.assertThat;
@Category(Flip6.class)
public class JobMasterTest extends TestLogger {
- private final Time testingTimeout = Time.seconds(10L);
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
- @Test
- public void testHeartbeatTimeoutWithTaskManager() throws Exception {
- final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices();
- final TestingLeaderRetrievalService rmLeaderRetrievalService = new TestingLeaderRetrievalService(
+ private static final Time testingTimeout = Time.seconds(10L);
+
+ private static final long heartbeatInterval = 1L;
+
+ private static final long heartbeatTimeout = 5L;
+
+ private static final JobGraph jobGraph = new JobGraph();
+
+ private static TestingRpcService rpcService;
+
+ private static HeartbeatServices fastHeartbeatServices;
+
+ private BlobServer blobServer;
+
+ private Configuration configuration;
+
+ private ResourceID jmResourceId;
+
+ private JobMasterId jobMasterId;
+
+ private TestingHighAvailabilityServices haServices;
+
+ private TestingLeaderRetrievalService rmLeaderRetrievalService;
+
+ private TestingFatalErrorHandler testingFatalErrorHandler;
+
+ @BeforeClass
+ public static void setupClass() {
+ rpcService = new TestingRpcService();
+
+ fastHeartbeatServices = new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout, rpcService.getScheduledExecutor());
+ }
+
+ @Before
+ public void setup() throws IOException {
+ configuration = new Configuration();
+ haServices = new TestingHighAvailabilityServices();
+ jobMasterId = JobMasterId.generate();
+ jmResourceId = ResourceID.generate();
+
+ testingFatalErrorHandler = new TestingFatalErrorHandler();
+
+ haServices.setCheckpointRecoveryFactory(new StandaloneCheckpointRecoveryFactory());
+
+ rmLeaderRetrievalService = new TestingLeaderRetrievalService(
null,
null);
haServices.setResourceManagerLeaderRetriever(rmLeaderRetrievalService);
- haServices.setCheckpointRecoveryFactory(new StandaloneCheckpointRecoveryFactory());
- final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
- final String jobManagerAddress = "jm";
- final JobMasterId jobMasterId = JobMasterId.generate();
- final ResourceID jmResourceId = new ResourceID(jobManagerAddress);
+ configuration.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+ blobServer = new BlobServer(configuration, new VoidBlobStore());
+
+ blobServer.start();
+ }
+
+ @After
+ public void teardown() throws Exception {
+ if (testingFatalErrorHandler != null) {
+ testingFatalErrorHandler.rethrowError();
+ }
- final String taskManagerAddress = "tm";
+ if (blobServer != null) {
+ blobServer.close();
+ }
+ }
+
+ @AfterClass
+ public static void teardownClass() {
+ if (rpcService != null) {
+ rpcService.stopService();
+ rpcService = null;
+ }
+ }
+
+ @Test
+ public void testHeartbeatTimeoutWithTaskManager() throws Exception {
final TaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
final TestingTaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGateway();
@@ -89,51 +157,40 @@ public class JobMasterTest extends TestLogger {
taskExecutorGateway.setHeartbeatJobManagerConsumer(heartbeatResourceIdFuture::complete);
taskExecutorGateway.setDisconnectJobManagerConsumer(tuple -> disconnectedJobManagerFuture.complete(tuple.f0));
- final TestingRpcService rpc = new TestingRpcService();
- rpc.registerGateway(taskManagerAddress, taskExecutorGateway);
-
- final long heartbeatInterval = 1L;
- final long heartbeatTimeout = 5L;
-
- final ScheduledExecutor scheduledExecutor = rpc.getScheduledExecutor();
- final HeartbeatServices heartbeatServices = new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout, scheduledExecutor);
-
- final JobGraph jobGraph = new JobGraph();
-
- Configuration configuration = new Configuration();
+ rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);
final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build();
final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
- try (BlobServer blobServer = new BlobServer(configuration, new VoidBlobStore())) {
- blobServer.start();
-
- final JobMaster jobMaster = new JobMaster(
- rpc,
- jobMasterConfiguration,
- jmResourceId,
- jobGraph,
- haServices,
- jobManagerSharedServices,
- heartbeatServices,
- blobServer,
- null,
- new NoOpOnCompletionActions(),
- testingFatalErrorHandler,
- FlinkUserCodeClassLoaders.parentFirst(new URL[0], JobMasterTest.class.getClassLoader()),
- null,
- null);
-
- CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
+ final JobMaster jobMaster = new JobMaster(
+ rpcService,
+ jobMasterConfiguration,
+ jmResourceId,
+ jobGraph,
+ haServices,
+ jobManagerSharedServices,
+ fastHeartbeatServices,
+ blobServer,
+ null,
+ new NoOpOnCompletionActions(),
+ testingFatalErrorHandler,
+ JobMasterTest.class.getClassLoader(),
+ null,
+ null);
+
+ CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
+ try {
// wait for the start to complete
startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
// register task manager will trigger monitor heartbeat target, schedule heartbeat request at interval time
- CompletableFuture<RegistrationResponse> registrationResponse = jobMasterGateway
- .registerTaskManager(taskManagerAddress, taskManagerLocation, testingTimeout);
+ CompletableFuture<RegistrationResponse> registrationResponse = jobMasterGateway.registerTaskManager(
+ taskExecutorGateway.getAddress(),
+ taskManagerLocation,
+ testingTimeout);
// wait for the completion of the registration
registrationResponse.get();
@@ -145,39 +202,17 @@ public class JobMasterTest extends TestLogger {
final JobID disconnectedJobManager = disconnectedJobManagerFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
assertThat(disconnectedJobManager, Matchers.equalTo(jobGraph.getJobID()));
-
- // check if a concurrent error occurred
- testingFatalErrorHandler.rethrowError();
-
} finally {
jobManagerSharedServices.shutdown();
- rpc.stopService();
+ RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
}
}
@Test
public void testHeartbeatTimeoutWithResourceManager() throws Exception {
final String resourceManagerAddress = "rm";
- final String jobManagerAddress = "jm";
final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
- final JobMasterId jobMasterId = JobMasterId.generate();
final ResourceID rmResourceId = new ResourceID(resourceManagerAddress);
- final ResourceID jmResourceId = new ResourceID(jobManagerAddress);
- final JobGraph jobGraph = new JobGraph();
-
- final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices();
- final TestingLeaderRetrievalService rmLeaderRetrievalService = new TestingLeaderRetrievalService(
- null,
- null);
- haServices.setResourceManagerLeaderRetriever(rmLeaderRetrievalService);
- haServices.setCheckpointRecoveryFactory(new StandaloneCheckpointRecoveryFactory());
-
- final TestingRpcService rpc = new TestingRpcService();
-
- final long heartbeatInterval = 1L;
- final long heartbeatTimeout = 5L;
- final ScheduledExecutor scheduledExecutor = rpc.getScheduledExecutor();
- final HeartbeatServices heartbeatServices = new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout, scheduledExecutor);
final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway(
resourceManagerId,
@@ -197,36 +232,30 @@ public class JobMasterTest extends TestLogger {
resourceManagerGateway.setDisconnectJobManagerConsumer(tuple -> disconnectedJobManagerFuture.complete(tuple.f0));
- rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
-
- final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
-
- Configuration configuration = new Configuration();
+ rpcService.registerGateway(resourceManagerAddress, resourceManagerGateway);
final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build();
final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
- try (BlobServer blobServer = new BlobServer(configuration, new VoidBlobStore())) {
- blobServer.start();
-
- final JobMaster jobMaster = new JobMaster(
- rpc,
- jobMasterConfiguration,
- jmResourceId,
- jobGraph,
- haServices,
- jobManagerSharedServices,
- heartbeatServices,
- blobServer,
- null,
- new NoOpOnCompletionActions(),
- testingFatalErrorHandler,
- FlinkUserCodeClassLoaders.parentFirst(new URL[0], JobMasterTest.class.getClassLoader()),
- null,
- null);
-
- CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
+ final JobMaster jobMaster = new JobMaster(
+ rpcService,
+ jobMasterConfiguration,
+ jmResourceId,
+ jobGraph,
+ haServices,
+ jobManagerSharedServices,
+ fastHeartbeatServices,
+ blobServer,
+ null,
+ new NoOpOnCompletionActions(),
+ testingFatalErrorHandler,
+ JobMasterTest.class.getClassLoader(),
+ null,
+ null);
+
+ CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
+ try {
// wait for the start operation to complete
startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
@@ -246,13 +275,9 @@ public class JobMasterTest extends TestLogger {
// heartbeat timeout should trigger disconnect JobManager from ResourceManager
assertThat(disconnectedJobManager, Matchers.equalTo(jobGraph.getJobID()));
-
- // check if a concurrent error occurred
- testingFatalErrorHandler.rethrowError();
-
} finally {
jobManagerSharedServices.shutdown();
- rpc.stopService();
+ RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
}
}
[3/9] flink git commit: [hotfix] [yarn] Remove unnecessary
TaskManager configuration generation
Posted by tr...@apache.org.
[hotfix] [yarn] Remove unnecessary TaskManager configuration generation
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/24c24297
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/24c24297
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/24c24297
Branch: refs/heads/master
Commit: 24c242977991c124d928798b4d71d569b6a32e4c
Parents: 4b89b5d
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Feb 7 18:58:32 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:53 2018 +0100
----------------------------------------------------------------------
.../apache/flink/yarn/YarnResourceManager.java | 24 ++++++++------------
1 file changed, 10 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/24c24297/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 305a2f5..d07ac5a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -20,9 +20,7 @@ package org.apache.flink.yarn;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
-import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -60,9 +58,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-
-import scala.concurrent.duration.FiniteDuration;
/**
* The yarn implementation of the resource manager. Used when the system is started
@@ -443,17 +438,18 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
taskManagerParameters.taskManagerTotalMemoryMB(),
taskManagerParameters.taskManagerHeapSizeMB(),
taskManagerParameters.taskManagerDirectMemoryLimitMB());
- int timeout = flinkConfig.getInteger(TaskManagerOptions.MAX_REGISTRATION_DURATION.key(),
- DEFAULT_TASK_MANAGER_REGISTRATION_DURATION);
- FiniteDuration teRegistrationTimeout = new FiniteDuration(timeout, TimeUnit.SECONDS);
- final Configuration taskManagerConfig = BootstrapTools.generateTaskManagerConfiguration(
- flinkConfig, "", 0, 1, teRegistrationTimeout);
- log.debug("TaskManager configuration: {}", taskManagerConfig);
+
+ log.debug("TaskManager configuration: {}", flinkConfig);
ContainerLaunchContext taskExecutorLaunchContext = Utils.createTaskExecutorContext(
- flinkConfig, yarnConfig, env,
- taskManagerParameters, taskManagerConfig,
- currDir, YarnTaskExecutorRunner.class, log);
+ flinkConfig,
+ yarnConfig,
+ env,
+ taskManagerParameters,
+ flinkConfig,
+ currDir,
+ YarnTaskExecutorRunner.class,
+ log);
// set a special environment variable to uniquely identify this container
taskExecutorLaunchContext.getEnvironment()
[7/9] flink git commit: [hotfix] Initialize FileSystem in
TaskManagerRunner#main
Posted by tr...@apache.org.
[hotfix] Initialize FileSystem in TaskManagerRunner#main
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d9d89ff1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d9d89ff1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d9d89ff1
Branch: refs/heads/master
Commit: d9d89ff1ea08e2053ef56bd256881f00616dd74f
Parents: ab8316f
Author: Till Rohrmann <tr...@apache.org>
Authored: Sat Feb 17 19:18:35 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sun Feb 18 10:12:54 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/taskexecutor/TaskManagerRunner.java | 9 +++++++++
1 file changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d9d89ff1/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 2fc796d..4cb1beb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -53,6 +54,7 @@ import akka.actor.ActorSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.net.InetAddress;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
@@ -233,6 +235,13 @@ public class TaskManagerRunner implements FatalErrorHandler {
final Configuration configuration = GlobalConfiguration.loadConfiguration(configDir);
+ try {
+ FileSystem.initialize(configuration);
+ } catch (IOException e) {
+ throw new IOException("Error while setting the default " +
+ "filesystem scheme from configuration.", e);
+ }
+
SecurityUtils.install(new SecurityConfiguration(configuration));
try {