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/15 10:43:05 UTC
[22/22] flink git commit: [hotfix] [flip6] Remove unnecessary timeout
from SlotPool
[hotfix] [flip6] Remove unnecessary timeout from SlotPool
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/341467dd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/341467dd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/341467dd
Branch: refs/heads/master
Commit: 341467dd58e4bdcfe133a31d8cb32b6a6a8d7559
Parents: 1321c17
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 13 13:18:01 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Feb 15 08:46:15 2018 +0100
----------------------------------------------------------------------
.../apache/flink/configuration/JobManagerOptions.java | 14 ++------------
.../org/apache/flink/runtime/jobmaster/JobMaster.java | 3 +--
.../flink/runtime/jobmaster/slotpool/SlotPool.java | 13 ++++++++-----
.../runtime/jobmaster/slotpool/SlotPoolRpcTest.java | 3 +--
.../runtime/jobmaster/slotpool/SlotPoolTest.java | 3 +--
5 files changed, 13 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/341467dd/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 a6df986..74ec5b7 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
@@ -135,22 +135,12 @@ public class JobManagerOptions {
public static final ConfigOption<Long> SLOT_REQUEST_TIMEOUT =
key("slot.request.timeout")
- .defaultValue(10 * 60 * 1000L)
+ .defaultValue(5L * 60L * 1000L)
.withDescription("The timeout in milliseconds for requesting a slot from Slot Pool.");
- public static final ConfigOption<Long> SLOT_REQUEST_RM_TIMEOUT =
- key("slot.request.resourcemanager.timeout")
- .defaultValue(10 * 1000L)
- .withDescription("The timeout in milliseconds for sending a request to Resource Manager.");
-
- public static final ConfigOption<Long> SLOT_ALLOCATION_RM_TIMEOUT =
- key("slot.allocation.resourcemanager.timeout")
- .defaultValue(5 * 60 * 1000L)
- .withDescription("The timeout in milliseconds for allocation a slot from Resource Manager.");
-
public static final ConfigOption<Long> SLOT_IDLE_TIMEOUT =
key("slot.idle.timeout")
- .defaultValue(5 * 60 * 1000L)
+ .defaultValue(20L * 1000L)
.withDescription("The timeout in milliseconds for a idle slot in Slot Pool.");
// ---------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/341467dd/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 6c5e524..8aa2617 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
@@ -280,8 +280,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
rpcService,
jobGraph.getJobID(),
SystemClock.getInstance(),
- Time.milliseconds(configuration.getLong(JobManagerOptions.SLOT_REQUEST_TIMEOUT)),
- rpcTimeout,
+ rpcTimeout, Time.milliseconds(configuration.getLong(JobManagerOptions.SLOT_REQUEST_TIMEOUT)),
Time.milliseconds(configuration.getLong(JobManagerOptions.SLOT_IDLE_TIMEOUT)));
this.slotPoolGateway = slotPool.getSelfGateway(SlotPoolGateway.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/341467dd/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 59f382f..72bb7e1 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
@@ -22,6 +22,7 @@ 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.JobManagerOptions;
+import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -137,8 +138,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
rpcService,
jobId,
SystemClock.getInstance(),
- Time.milliseconds(JobManagerOptions.SLOT_ALLOCATION_RM_TIMEOUT.defaultValue()),
- Time.milliseconds(JobManagerOptions.SLOT_REQUEST_RM_TIMEOUT.defaultValue()),
+ AkkaUtils.getDefaultTimeout(),
+ Time.milliseconds(JobManagerOptions.SLOT_REQUEST_TIMEOUT.defaultValue()),
Time.milliseconds(JobManagerOptions.SLOT_IDLE_TIMEOUT.defaultValue()));
}
@@ -146,8 +147,8 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
RpcService rpcService,
JobID jobId,
Clock clock,
- Time slotRequestTimeout,
Time rpcTimeout,
+ Time slotRequestTimeout,
Time idleSlotTimeout) {
super(rpcService);
@@ -714,7 +715,9 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
private void checkTimeoutSlotAllocation(SlotRequestId slotRequestID) {
PendingRequest request = pendingRequests.removeKeyA(slotRequestID);
if (request != null) {
- failPendingRequest(request, new TimeoutException("Slot allocation request " + slotRequestID + " timed out"));
+ failPendingRequest(
+ request,
+ new TimeoutException("Slot allocation request " + slotRequestID + " timed out"));
}
}
@@ -733,7 +736,7 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
if (request != null) {
failPendingRequest(
request,
- new NoResourceAvailableException("No slot available and no connection to Resource Manager established."));
+ new TimeoutException("No slot available and no connection to Resource Manager established."));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/341467dd/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 9962250..bf8037d 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
@@ -360,8 +360,7 @@ public class SlotPoolRpcTest extends TestLogger {
rpcService,
jobId,
clock,
- slotRequestTimeout,
- rpcTimeout,
+ rpcTimeout, slotRequestTimeout,
idleSlotTimeout);
releaseSlotConsumer = null;
http://git-wip-us.apache.org/repos/asf/flink/blob/341467dd/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 0c0b42c..6a85461 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
@@ -587,8 +587,7 @@ public class SlotPoolTest extends TestLogger {
rpcService,
jobId,
clock,
- TestingUtils.infiniteTime(),
- TestingUtils.infiniteTime(),
+ TestingUtils.infiniteTime(), TestingUtils.infiniteTime(),
timeout);
try {