You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tillrohrmann <gi...@git.apache.org> on 2018/03/21 17:48:24 UTC
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
GitHub user tillrohrmann opened a pull request:
https://github.com/apache/flink/pull/5739
[FLINK-9047] Fix slot recycling in case of failed release
## What is the purpose of the change
In case that a slot cannot be released it will only recycled/reused if the owning
TaskExecutor is still registered at the SlotPool. If this is not the case then we
drop the slot from the SlotPool.
cc @GJL
## Brief change log
- Only recycle slots which could not be released if owner is still registered
## Verifying this change
- Added `SlotPoolTest#testReleasingIdleSlotFailed`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tillrohrmann/flink moreLogging
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5739.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5739
----
commit fbaa3c1c94af6d7ba26ec6e554e35d9d9b400054
Author: Till Rohrmann <tr...@...>
Date: 2018-03-21T14:09:46Z
[hotfix] Improve Flip-6 component logging
commit 46cda67300fd153828769af8b4139b64b60e34d4
Author: Till Rohrmann <tr...@...>
Date: 2018-03-21T15:48:52Z
[FLINK-9047] Fix slot recycling in case of failed release
In case that a slot cannot be released it will only recycled/reused if the owning
TaskExecutor is still registered at the SlotPool. If this is not the case then we
drop the slot from the SlotPool.
commit c9f3e037021dc8c37eceaaac181eb19c141f0bb7
Author: Till Rohrmann <tr...@...>
Date: 2018-03-21T17:44:25Z
[hotfix] Remove unused method from SlotPool
----
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176668855
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java ---
@@ -634,6 +653,99 @@ public void testCheckIdleSlot() throws Exception {
}
}
+ /**
+ * Tests that idle slots which cannot be released are only recycled if the owning {@link TaskExecutor}
+ * is still registered at the {@link SlotPool}. See FLINK-9047.
+ */
+ @Test
+ public void testReleasingIdleSlotFailed() throws Exception {
+ final ManualClock clock = new ManualClock();
+ final SlotPool slotPool = new SlotPool(
+ rpcService,
+ jobId,
+ clock,
+ TestingUtils.infiniteTime(),
+ timeout);
+
+ try {
+ final SlotPoolGateway slotPoolGateway = setupSlotPool(slotPool, resourceManagerGateway);
+
+ final AllocationID expiredAllocationId = new AllocationID();
+ final SlotOffer slotToExpire = new SlotOffer(expiredAllocationId, 0, ResourceProfile.UNKNOWN);
+
+ final ArrayDeque<CompletableFuture<Acknowledge>> responseQueue = new ArrayDeque<>(2);
+ taskManagerGateway.setFreeSlotFunction((AllocationID allocationId, Throwable cause) -> {
+ if (responseQueue.isEmpty()) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ } else {
+ return responseQueue.pop();
+ }
+ });
+
+ responseQueue.add(FutureUtils.completedExceptionally(new FlinkException("Test failure")));
+
+ final CompletableFuture<Acknowledge> responseFuture = new CompletableFuture<>();
+ responseQueue.add(responseFuture);
+
+ assertThat(
+ slotPoolGateway.registerTaskManager(taskManagerLocation.getResourceID()).get(),
+ Matchers.is(Acknowledge.get()));
+
+ assertThat(
+ slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire).get(),
+ Matchers.is(true));
+
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ CompletableFuture<LogicalSlot> allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // wait until the slot has been fulfilled with the previously idling slot
+ final LogicalSlot logicalSlot = allocatedSlotFuture.get();
+ assertThat(logicalSlot.getAllocationId(), Matchers.is(expiredAllocationId));
+
+ // return the slot
+ slotPool.getSlotOwner().returnAllocatedSlot(logicalSlot).get();
+
+ // advance the time so that the returned slot is now idling
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ // request a new slot after the idling slot has been released
+ allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // release the TaskExecutor before we get a response from the slot releasing
+ slotPoolGateway.releaseTaskManager(taskManagerLocation.getResourceID()).get();
+
+ // let the slot releasing fail --> since there the owning TaskExecutor is no longer registered
+ // the slot should be discarded
+ responseFuture.completeExceptionally(new FlinkException("Second test exception"));
+
+ try {
+ // since the slot msut have been discarded, we cannot fulfill the slot request
--- End diff --
Good catch
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176401066
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java ---
@@ -634,6 +653,99 @@ public void testCheckIdleSlot() throws Exception {
}
}
+ /**
+ * Tests that idle slots which cannot be released are only recycled if the owning {@link TaskExecutor}
+ * is still registered at the {@link SlotPool}. See FLINK-9047.
+ */
+ @Test
+ public void testReleasingIdleSlotFailed() throws Exception {
+ final ManualClock clock = new ManualClock();
+ final SlotPool slotPool = new SlotPool(
+ rpcService,
+ jobId,
+ clock,
+ TestingUtils.infiniteTime(),
+ timeout);
+
+ try {
+ final SlotPoolGateway slotPoolGateway = setupSlotPool(slotPool, resourceManagerGateway);
+
+ final AllocationID expiredAllocationId = new AllocationID();
+ final SlotOffer slotToExpire = new SlotOffer(expiredAllocationId, 0, ResourceProfile.UNKNOWN);
+
+ final ArrayDeque<CompletableFuture<Acknowledge>> responseQueue = new ArrayDeque<>(2);
+ taskManagerGateway.setFreeSlotFunction((AllocationID allocationId, Throwable cause) -> {
+ if (responseQueue.isEmpty()) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ } else {
+ return responseQueue.pop();
+ }
+ });
+
+ responseQueue.add(FutureUtils.completedExceptionally(new FlinkException("Test failure")));
+
+ final CompletableFuture<Acknowledge> responseFuture = new CompletableFuture<>();
+ responseQueue.add(responseFuture);
+
+ assertThat(
+ slotPoolGateway.registerTaskManager(taskManagerLocation.getResourceID()).get(),
+ Matchers.is(Acknowledge.get()));
+
+ assertThat(
+ slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire).get(),
+ Matchers.is(true));
+
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ CompletableFuture<LogicalSlot> allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // wait until the slot has been fulfilled with the previously idling slot
+ final LogicalSlot logicalSlot = allocatedSlotFuture.get();
+ assertThat(logicalSlot.getAllocationId(), Matchers.is(expiredAllocationId));
+
+ // return the slot
+ slotPool.getSlotOwner().returnAllocatedSlot(logicalSlot).get();
+
+ // advance the time so that the returned slot is now idling
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ // request a new slot after the idling slot has been released
+ allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // release the TaskExecutor before we get a response from the slot releasing
+ slotPoolGateway.releaseTaskManager(taskManagerLocation.getResourceID()).get();
+
+ // let the slot releasing fail --> since there the owning TaskExecutor is no longer registered
+ // the slot should be discarded
+ responseFuture.completeExceptionally(new FlinkException("Second test exception"));
+
+ try {
+ // since the slot msut have been discarded, we cannot fulfill the slot request
--- End diff --
nit: *msut*
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5739
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176404156
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/TaskManagerSlot.java ---
@@ -139,6 +139,16 @@ public boolean isMatchingRequirement(ResourceProfile required) {
return resourceProfile.isMatching(required);
}
+ @Override
+ public String toString() {
+ return "TaskManagerSlot{" +
--- End diff --
`resourceProfile` and `taskManagerConnection` are deliberately not included?
---
[GitHub] flink issue #5739: [FLINK-9047] Fix slot recycling in case of failed release
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/5739
Thanks for the review @GJL. Merging this PR.
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176668534
--- Diff: flink-runtime/src/test/resources/log4j-test.properties ---
@@ -16,7 +16,7 @@
# limitations under the License.
################################################################################
-log4j.rootLogger=OFF, console
+log4j.rootLogger=DEBUG, console
--- End diff --
Good catch. Will fix it.
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176668568
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java ---
@@ -634,6 +653,99 @@ public void testCheckIdleSlot() throws Exception {
}
}
+ /**
+ * Tests that idle slots which cannot be released are only recycled if the owning {@link TaskExecutor}
+ * is still registered at the {@link SlotPool}. See FLINK-9047.
+ */
+ @Test
+ public void testReleasingIdleSlotFailed() throws Exception {
+ final ManualClock clock = new ManualClock();
+ final SlotPool slotPool = new SlotPool(
+ rpcService,
+ jobId,
+ clock,
+ TestingUtils.infiniteTime(),
+ timeout);
+
+ try {
+ final SlotPoolGateway slotPoolGateway = setupSlotPool(slotPool, resourceManagerGateway);
+
+ final AllocationID expiredAllocationId = new AllocationID();
+ final SlotOffer slotToExpire = new SlotOffer(expiredAllocationId, 0, ResourceProfile.UNKNOWN);
+
+ final ArrayDeque<CompletableFuture<Acknowledge>> responseQueue = new ArrayDeque<>(2);
+ taskManagerGateway.setFreeSlotFunction((AllocationID allocationId, Throwable cause) -> {
+ if (responseQueue.isEmpty()) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ } else {
+ return responseQueue.pop();
+ }
+ });
+
+ responseQueue.add(FutureUtils.completedExceptionally(new FlinkException("Test failure")));
+
+ final CompletableFuture<Acknowledge> responseFuture = new CompletableFuture<>();
+ responseQueue.add(responseFuture);
+
+ assertThat(
+ slotPoolGateway.registerTaskManager(taskManagerLocation.getResourceID()).get(),
+ Matchers.is(Acknowledge.get()));
+
+ assertThat(
+ slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire).get(),
+ Matchers.is(true));
+
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ CompletableFuture<LogicalSlot> allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // wait until the slot has been fulfilled with the previously idling slot
+ final LogicalSlot logicalSlot = allocatedSlotFuture.get();
+ assertThat(logicalSlot.getAllocationId(), Matchers.is(expiredAllocationId));
+
+ // return the slot
+ slotPool.getSlotOwner().returnAllocatedSlot(logicalSlot).get();
+
+ // advance the time so that the returned slot is now idling
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ // request a new slot after the idling slot has been released
+ allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // release the TaskExecutor before we get a response from the slot releasing
+ slotPoolGateway.releaseTaskManager(taskManagerLocation.getResourceID()).get();
+
+ // let the slot releasing fail --> since there the owning TaskExecutor is no longer registered
--- End diff --
Will fix it.
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176347600
--- Diff: flink-runtime/src/test/resources/log4j-test.properties ---
@@ -16,7 +16,7 @@
# limitations under the License.
################################################################################
-log4j.rootLogger=OFF, console
+log4j.rootLogger=DEBUG, console
--- End diff --
I think this should be reverted.
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176668913
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/TaskManagerSlot.java ---
@@ -139,6 +139,16 @@ public boolean isMatchingRequirement(ResourceProfile required) {
return resourceProfile.isMatching(required);
}
+ @Override
+ public String toString() {
+ return "TaskManagerSlot{" +
--- End diff --
Will add them.
---
[GitHub] flink pull request #5739: [FLINK-9047] Fix slot recycling in case of failed ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5739#discussion_r176400741
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java ---
@@ -634,6 +653,99 @@ public void testCheckIdleSlot() throws Exception {
}
}
+ /**
+ * Tests that idle slots which cannot be released are only recycled if the owning {@link TaskExecutor}
+ * is still registered at the {@link SlotPool}. See FLINK-9047.
+ */
+ @Test
+ public void testReleasingIdleSlotFailed() throws Exception {
+ final ManualClock clock = new ManualClock();
+ final SlotPool slotPool = new SlotPool(
+ rpcService,
+ jobId,
+ clock,
+ TestingUtils.infiniteTime(),
+ timeout);
+
+ try {
+ final SlotPoolGateway slotPoolGateway = setupSlotPool(slotPool, resourceManagerGateway);
+
+ final AllocationID expiredAllocationId = new AllocationID();
+ final SlotOffer slotToExpire = new SlotOffer(expiredAllocationId, 0, ResourceProfile.UNKNOWN);
+
+ final ArrayDeque<CompletableFuture<Acknowledge>> responseQueue = new ArrayDeque<>(2);
+ taskManagerGateway.setFreeSlotFunction((AllocationID allocationId, Throwable cause) -> {
+ if (responseQueue.isEmpty()) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ } else {
+ return responseQueue.pop();
+ }
+ });
+
+ responseQueue.add(FutureUtils.completedExceptionally(new FlinkException("Test failure")));
+
+ final CompletableFuture<Acknowledge> responseFuture = new CompletableFuture<>();
+ responseQueue.add(responseFuture);
+
+ assertThat(
+ slotPoolGateway.registerTaskManager(taskManagerLocation.getResourceID()).get(),
+ Matchers.is(Acknowledge.get()));
+
+ assertThat(
+ slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire).get(),
+ Matchers.is(true));
+
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ CompletableFuture<LogicalSlot> allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // wait until the slot has been fulfilled with the previously idling slot
+ final LogicalSlot logicalSlot = allocatedSlotFuture.get();
+ assertThat(logicalSlot.getAllocationId(), Matchers.is(expiredAllocationId));
+
+ // return the slot
+ slotPool.getSlotOwner().returnAllocatedSlot(logicalSlot).get();
+
+ // advance the time so that the returned slot is now idling
+ clock.advanceTime(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ slotPool.triggerCheckIdleSlot();
+
+ // request a new slot after the idling slot has been released
+ allocatedSlotFuture = slotPoolGateway.allocateSlot(
+ new SlotRequestId(),
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ true,
+ timeout);
+
+ // release the TaskExecutor before we get a response from the slot releasing
+ slotPoolGateway.releaseTaskManager(taskManagerLocation.getResourceID()).get();
+
+ // let the slot releasing fail --> since there the owning TaskExecutor is no longer registered
--- End diff --
nit: *[...] since there the owning [...]*
---