You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/05/20 08:54:02 UTC
[GitHub] [flink] zentol opened a new pull request #12264: [FLINK-17668][netty] Release partitions asynchronously
zentol opened a new pull request #12264:
URL: https://github.com/apache/flink/pull/12264
With this PR the local release of partitions in the `NettyShuffleEnvironment` is executed asynchronously. This primarily includes the deletion of files. We have seen instances where this operation was taking a long time, blocking the TaskExecutor main thread.
For this we are re-using the existing `ioExecutor` from the `TaskExecutor` and also pass it into the `NettyShuffleEnvironment`. This executor was so far only used for handling log file requests.
The number of threads of this executor was increased from 1 -> 2.
Given that at most 1 thread should be busy handling log file requests (?), we should at least 1 thread processing the partition releases, which should be sufficient as we so far only used 1 thread as well (a busy one at that).
We still want these partitions to be cleaned up in a timely fashion; not for correctness, but to reduce disk usage.
There is a new (hidden) option for increasing this thread count, but ideally users should never have to use it.
There are ultimately 3 places where we could introduce the asynchronous behavior; in the TaskExecutor, PartitionTracker or ShuffleEnvironment.
Doing this in the TaskExecutor would imply doing all operations on the PartitionTracker with the ioExecutor, since the tracker requires single-threaded access. This would add complexity where we want it the least, and could easily lead to bugs since we start switching back and forth between executors.
The PartitionTracker is a good candidate; it can selectively use the executor for the release call into the ShuffleEnvironment, and would guard the TaskExecutor against such blocking calls regardless of ShuffleEnvironment implementations. So far the tracker was not aware of any asynchronous stuff however, and there may be value in keeping it that way.
For this PR I opted for the ShuffleEnvironment. It seems reasonable that the local release should be a non-blocking operation; any ShuffleEnvironment that communicates to the outside will necessarily have to do some non-blocking operations, and being provided an Executor through the ShuffleEnvironmentContext may make this easier, although it has the downside that the ShuffleEnvironment may over-use it and potentially deny other other usages of the executor. On the flip-side, this implies that we only use the executor when necessary; for a ShuffleEnvironment that implements the local release in a non-blocking fashion the safeguard in the PartitionTracker would just be a waste of resources.
One thing left to do is adding a test to ensure a blocking partition release is not blocking the TaskExecutor main thread.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r430381100
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##########
@@ -490,6 +490,13 @@
+ " size will be used. The exact size of JVM Overhead can be explicitly specified by setting the min/max"
+ " size to the same value.");
+ @Documentation.ExcludeFromDocumentation("This option just serves as a last-ditch escape hatch.")
+ public static final ConfigOption<Integer> NUM_IO_THREADS =
+ key("taskmanager.io.threads.num")
+ .intType()
+ .defaultValue(2)
+ .withDescription("The number of threads to use for non-critical IO operations.");
Review comment:
The `HighAvailabilityServices` use the `ioExecutor` for I/O tasks. Concretely, they use it do dispose a completed checkpoint. In that sense it is also an I/O operation and I think they could be served by the same executor. But in order to keep the scope smaller, we don't have to do it in this PR.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
##########
@@ -265,10 +265,15 @@ public static TaskManagerServices fromConfiguration(
// start the I/O manager, it will create some temp directories.
final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
+ final ExecutorService ioExecutor = Executors.newFixedThreadPool(
Review comment:
I think we should really unify them. `taskIOExecutor` is effectively the same as `ioExecutor` because it is used by the `TaskExecutorLocalStateStoresManager` for discarding local state. I think it is better to use common thread pools until we have a really good reason for separating thread pools. Creating more thread pools will also increase the required resource foot print.
Concerning the number of threads, I would suggest to use `ClusterOptions.CLUSTER_IO_EXECUTOR_POOL_SIZE` which defaults to the number of available cores if it has not been specified. This will also allow users to increase the thread pool if things take too long with a single thread.
##########
File path: flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/executor/TestExecutorServiceResource.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.testutils.executor;
+
+import org.junit.rules.ExternalResource;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+
+/**
+ * Resource which starts an {@link ExecutorService} for testing purposes.
+ */
+public class TestExecutorServiceResource extends ExternalResource {
+
+ private final Supplier<ExecutorService> serviceFactory;
+
+ private ExecutorService executorService;
+
+ public TestExecutorServiceResource(Supplier<ExecutorService> serviceFactory) {
+ this.serviceFactory = serviceFactory;
+ }
+
+ @Override
+ protected void before() throws Throwable {
+ executorService = serviceFactory.get();
+ }
+
+ public ExecutorService getExecutorService() {
+ return executorService;
+ }
+
+ @Override
+ protected void after() {
+ if (executorService != null) {
+ executorService.shutdown();
Review comment:
I would suggest to use `ExecutorUtils.gracefulShutdown`.
##########
File path: flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/executor/TestExecutorServiceResource.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.testutils.executor;
+
+import org.junit.rules.ExternalResource;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+
+/**
+ * Resource which starts an {@link ExecutorService} for testing purposes.
+ */
+public class TestExecutorServiceResource extends ExternalResource {
Review comment:
This class seems to duplicate `TestingScheduledExecutor`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
}, {
"hash" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 59111304bfc04b61f87daae0fd5c3f46fc3ef254 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288)
* 8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r431066056
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
##########
@@ -270,7 +271,8 @@ public static TaskManagerServicesConfiguration fromConfiguration(
final String[] alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(configuration);
- final int numIoThreads = configuration.get(TaskManagerOptions.NUM_IO_THREADS);
+ // multiply core-count to be on the safer side, since we used a pool with size=64 in the past
+ final int numIoThreads = ClusterEntrypointUtils.getPoolSize(configuration) * 4;
Review comment:
```suggestion
final int numIoThreads = ClusterEntrypointUtils.getPoolSize(configuration);
```
I will update `getPoolSize` to return the new default value of `4 * cores` as part of changing the type of the thread pool on the JM side.
##########
File path: flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/executor/TestExecutorResource.java
##########
@@ -19,35 +19,37 @@
import org.junit.rules.ExternalResource;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
/**
- * Resource which starts an {@link ExecutorService} for testing purposes.
+ * Resource which starts/stops an {@link ExecutorService} for testing purposes.
*/
-public class TestExecutorServiceResource extends ExternalResource {
+public class TestExecutorResource extends ExternalResource {
private final Supplier<ExecutorService> serviceFactory;
- private ExecutorService executorService;
+ private ExecutorService executor;
Review comment:
The name `executorService` could have been kept.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##########
@@ -366,11 +367,15 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ final ExecutorService ioExecutor = Executors.newCachedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
Review comment:
This could be moved into the `fromConfiguration` method. Given that the `TaskManagerServices` is responsible for managing the created `ExecutorService`, I think it is fine to move the creation into the `fromConfiguration` method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 3510bfd56ae6a431783bbade1881dd967b271457 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139)
* 084e28079e239e9485c36f843348323c9a81ad6a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r429896972
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##########
@@ -490,6 +490,13 @@
+ " size will be used. The exact size of JVM Overhead can be explicitly specified by setting the min/max"
+ " size to the same value.");
+ @Documentation.ExcludeFromDocumentation("This option just serves as a last-ditch escape hatch.")
+ public static final ConfigOption<Integer> NUM_IO_THREADS =
+ key("taskmanager.io.threads.num")
+ .intType()
+ .defaultValue(2)
+ .withDescription("The number of threads to use for non-critical IO operations.");
Review comment:
hmm...maybe; the use-cases are quite different though?
`CLUSTER_IO_EXECUTOR_POOL_SIZE` is currently used for the pool we use for the HA executor pool, which is probably pretty important stuff.
Meanwhile, on the TM it is mostly about disk cleanup.
Should these really be controlled through the same option? If a user sees a problem with slow disk cleanup and increases the pool size, should this really affect the HA side?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r431091900
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##########
@@ -366,11 +367,15 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ final ExecutorService ioExecutor = Executors.newCachedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
Review comment:
But we can also keep it as is. The important aspect is that everyone knows that `fromConfiguration` takes ownership over the passed `ioExecutor`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934)
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* eafbd98c812227cb7d9ce7158de1a23309855509 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
}, {
"hash" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303",
"triggerID" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"triggerType" : "PUSH"
}, {
"hash" : "c5dcd45e890a076e0a66d4e6e24fdab192e3a724",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c5dcd45e890a076e0a66d4e6e24fdab192e3a724",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303)
* c5dcd45e890a076e0a66d4e6e24fdab192e3a724 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r430897493
##########
File path: flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/executor/TestExecutorServiceResource.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.testutils.executor;
+
+import org.junit.rules.ExternalResource;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+
+/**
+ * Resource which starts an {@link ExecutorService} for testing purposes.
+ */
+public class TestExecutorServiceResource extends ExternalResource {
Review comment:
Can we unify them and get rid of one of them?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* eafbd98c812227cb7d9ce7158de1a23309855509 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948)
* 3510bfd56ae6a431783bbade1881dd967b271457 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 59111304bfc04b61f87daae0fd5c3f46fc3ef254 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 084e28079e239e9485c36f843348323c9a81ad6a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273)
* 59111304bfc04b61f87daae0fd5c3f46fc3ef254 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r428047035
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
##########
@@ -280,7 +273,65 @@ public void testClusterPartitionRelease() throws Exception {
);
}
- private <C> void testPartitionRelease(PartitionTrackerSetup<C> partitionTrackerSetup, TestAction<C> testAction) throws Exception {
+ @Test
+ public void testBlockingLocalPartitionReleaseDoesNotBlockTaskExecutor() throws Exception {
+ BlockerSync sync = new BlockerSync();
+ ResultPartitionManager blockingResultPartitionManager = new ResultPartitionManager() {
+ @Override
+ public void releasePartition(ResultPartitionID partitionId, Throwable cause) {
+ sync.blockNonInterruptible();
+ super.releasePartition(partitionId, cause);
+ }
+ };
+
+ NettyShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder()
+ .setResultPartitionManager(blockingResultPartitionManager)
+ .setIoExecutor(java.util.concurrent.Executors.newFixedThreadPool(1))
Review comment:
I would suggest to also shut this executor service down at the end of the test. It might be necessary to unblock the release operation for this.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##########
@@ -490,6 +490,13 @@
+ " size will be used. The exact size of JVM Overhead can be explicitly specified by setting the min/max"
+ " size to the same value.");
+ @Documentation.ExcludeFromDocumentation("This option just serves as a last-ditch escape hatch.")
+ public static final ConfigOption<Integer> NUM_IO_THREADS =
+ key("taskmanager.io.threads.num")
+ .intType()
+ .defaultValue(2)
+ .withDescription("The number of threads to use for non-critical IO operations.");
Review comment:
We might be able to unify this configuration option with `ClusterOptions.CLUSTER_IO_EXECUTOR_POOL_SIZE`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
##########
@@ -265,10 +265,15 @@ public static TaskManagerServices fromConfiguration(
// start the I/O manager, it will create some temp directories.
final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
+ final ExecutorService ioExecutor = Executors.newFixedThreadPool(
Review comment:
Can the `ioExecutor` also replace the `taskIOExecutor`?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
##########
@@ -100,6 +105,27 @@ public void testRegisterTaskWithInsufficientBuffers() throws Exception {
testRegisterTaskWithLimitedBuffers(bufferCount);
}
+ @Test
+ public void testSlowIODoesNotBlockRelease() throws Exception {
+ BlockerSync sync = new BlockerSync();
Review comment:
I guess a `OneShotLatch` would also work here if the test threads call the trigger on it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12264: [FLINK-17668][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* eafbd98c812227cb7d9ce7158de1a23309855509 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r429893280
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
##########
@@ -100,6 +105,27 @@ public void testRegisterTaskWithInsufficientBuffers() throws Exception {
testRegisterTaskWithLimitedBuffers(bufferCount);
}
+ @Test
+ public void testSlowIODoesNotBlockRelease() throws Exception {
+ BlockerSync sync = new BlockerSync();
Review comment:
I used a BlockerSync since it allows blocking without throwing checked exceptions.
`OneShotLatch#await` throwing `InterruptedException` is a bit annoying. Maybe we should change that; for testing purposes it seems irrelevant?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
}, {
"hash" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303",
"triggerID" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r431071230
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##########
@@ -366,11 +367,15 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ final ExecutorService ioExecutor = Executors.newCachedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
Review comment:
initially that was the plan, but the `TaskExecutorLocalStateStoresManagerTest` passes in a `Executors.directExecutor()` into `TaskManagerServices.fromConfiguration`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934)
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* eafbd98c812227cb7d9ce7158de1a23309855509 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
}, {
"hash" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303",
"triggerID" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 59111304bfc04b61f87daae0fd5c3f46fc3ef254 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288)
* 8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 3510bfd56ae6a431783bbade1881dd967b271457 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139)
* 084e28079e239e9485c36f843348323c9a81ad6a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r430897263
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
##########
@@ -265,10 +265,15 @@ public static TaskManagerServices fromConfiguration(
// start the I/O manager, it will create some temp directories.
final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
+ final ExecutorService ioExecutor = Executors.newFixedThreadPool(
Review comment:
I think it is actually a good idea to move it away from the `RpcService's` executor because this is the same pool which is used to run the `RpcEndpoints`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 3510bfd56ae6a431783bbade1881dd967b271457 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r431074631
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##########
@@ -366,11 +367,15 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ final ExecutorService ioExecutor = Executors.newCachedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
Review comment:
although this doesn't seem related to the correctness of the test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r430548944
##########
File path: flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/executor/TestExecutorServiceResource.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.testutils.executor;
+
+import org.junit.rules.ExternalResource;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+
+/**
+ * Resource which starts an {@link ExecutorService} for testing purposes.
+ */
+public class TestExecutorServiceResource extends ExternalResource {
Review comment:
I wouldn't say that it is a duplicate since one works against the java `ExecutorService` while the other against the flink-runtime `ScheduledExecutor`.
They are ultimately similar yes, but this one is usable without depending on flink-runtime (in particular, without depending on the flink-runtime test-jar).
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
##########
@@ -265,10 +265,15 @@ public static TaskManagerServices fromConfiguration(
// start the I/O manager, it will create some temp directories.
final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
+ final ExecutorService ioExecutor = Executors.newFixedThreadPool(
Review comment:
Well, one reason is uncertainty.
Do you know how many resources the state cleanup currently siphons from the [RpcService executor](https://github.com/apache/flink/blob/5dededa06592e4b68023bb3a3282849038f18fa2/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java#L373)?
I don't disagree with your take in general, but as is the PR is pretty safe to back-port to older versions.
I'm not so sure about this if we include the taskIOExecutor change; we don't have to look back very far for issues when we moved things away from the RPC executor (FLINK-17248).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
}, {
"hash" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303",
"triggerID" : "8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc",
"triggerType" : "PUSH"
}, {
"hash" : "c5dcd45e890a076e0a66d4e6e24fdab192e3a724",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2344",
"triggerID" : "c5dcd45e890a076e0a66d4e6e24fdab192e3a724",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 8cf00b9f6fd8b76256883eedbdb8e79dea3c35dc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2303)
* c5dcd45e890a076e0a66d4e6e24fdab192e3a724 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2344)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12264: [FLINK-17668][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631339639
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 19c5f57b94cc56b70002031618c32d9e6f68effb (Wed May 20 08:56:08 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r431090914
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##########
@@ -366,11 +367,15 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ final ExecutorService ioExecutor = Executors.newCachedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
Review comment:
Jup. It is also a bit odd that `TaskExecutorLocalStateStoresManagerTest.testCreationFromConfig` needs to create a `TaskManagerServices` instance in order to test the `TaskExecutorLocalStateStoresManager`. I think one could test the same if one factored the creation of the `TaskExecutorLocalStateStoresManager` out into a separate method which is called by `TaskManagerServices.fromConfiguration()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #12264:
URL: https://github.com/apache/flink/pull/12264#discussion_r429894695
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
##########
@@ -265,10 +265,15 @@ public static TaskManagerServices fromConfiguration(
// start the I/O manager, it will create some temp directories.
final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
+ final ExecutorService ioExecutor = Executors.newFixedThreadPool(
Review comment:
Technically yes, but I wouldn't know how many threads the executor should have. It is also easier to judge how many you'd need if the task and TM operations are separated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* 9dbaf3094c0942b96a01060aba9d4ffbad9d1857 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934)
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* eafbd98c812227cb7d9ce7158de1a23309855509 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948)
* 3510bfd56ae6a431783bbade1881dd967b271457 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12264:
URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "19c5f57b94cc56b70002031618c32d9e6f68effb",
"triggerType" : "PUSH"
}, {
"hash" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1934",
"triggerID" : "9dbaf3094c0942b96a01060aba9d4ffbad9d1857",
"triggerType" : "PUSH"
}, {
"hash" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bb313e40f5a72dbf20cd0a8b48267063fd4f00af",
"triggerType" : "PUSH"
}, {
"hash" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=1948",
"triggerID" : "eafbd98c812227cb7d9ce7158de1a23309855509",
"triggerType" : "PUSH"
}, {
"hash" : "3510bfd56ae6a431783bbade1881dd967b271457",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139",
"triggerID" : "3510bfd56ae6a431783bbade1881dd967b271457",
"triggerType" : "PUSH"
}, {
"hash" : "084e28079e239e9485c36f843348323c9a81ad6a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273",
"triggerID" : "084e28079e239e9485c36f843348323c9a81ad6a",
"triggerType" : "PUSH"
}, {
"hash" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288",
"triggerID" : "59111304bfc04b61f87daae0fd5c3f46fc3ef254",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN
* bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN
* 084e28079e239e9485c36f843348323c9a81ad6a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2273)
* 59111304bfc04b61f87daae0fd5c3f46fc3ef254 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2288)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org