You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by uce <gi...@git.apache.org> on 2015/05/21 12:20:18 UTC
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
GitHub user uce opened a pull request:
https://github.com/apache/flink/pull/705
[FLINK-1954] [FLINK-1636] [runtime] Improve partition not found error handling
**Problem**: cancelling of tasks sometimes leads to misleading error messages about "not found partitions". This is an artifact of task cancelling. If a task (consumer) consumes data from another remote task (producer), its sends a partition request over the network. If the producer fails concurrently with this
request, the request returns with a PartitioNotFoundException to the consumer. If this error message is received *before* the consumer is cancelled (as a result of the failing producer), you see the misleading error being attributed to the consumer. This makes it hard to trace the root cause of the problem (the
failing producer).
**Solution**: when a consumer receives a remote PartitionNotFoundException, it asks the central job manager whether the producer is still running or has failed.
If the producer is still running, the partition request is send again (using an exponential back off with default max back off of 3s). If the following requests fail again, the consumer fails with a PartitionNotFoundException.
If the producer has failed, the consumer is cancelled.
If the producer is not running and has not failed, there is a bug either in the consumer task setup (e.g. requesting a non-existing result) or in the network stack (e.g. unsafe publication of produced results), in which case the error is attributed to the consumer.
---
The new Akka messages introduced with this change are only exchanged in error cases and don't affect normal operation.
Normal operation (not affected by this change):
```
TM1 => TM2: request result
TM2 => TM1: result
```
Error case:
```
TM1=>TM2: request result
TM2=>TM1: PartitionNotFoundException
TM1=>JM: check partition state
JM=>TM1: retrigger request -OR- cancel consumer
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/uce/incubator-flink partition_not_found-1636
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/705.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 #705
----
commit a480a638056e4f5300573f543acf46cd239f2674
Author: Ufuk Celebi <uc...@apache.org>
Date: 2015-05-11T14:34:55Z
[FLINK-1954] [FLINK-1636] [runtime] Improve partition not found error handling
Problem: cancelling of tasks sometimes leads to misleading error messages about
"not found partitions". This is an artifact of task cancelling. If a task
(consumer) consumes data from another remote task (producer), its sends a
partition request over the network. If the producer fails concurrently with this
request, the request returns with a PartitioNotFoundException to the consumer.
If this error message is received *before* the consumer is cancelled (as a
result of the failing producer), you see the misleading error being attributed
to the consumer. This makes it hard to trace the root cause of the problem (the
failing producer).
Solution: when a consumer receives a remote PartitionNotFoundException, it asks
the central job manager whether the producer is still running or has failed.
If the producer is still running, the partition request is send again (using an
exponential back off). If the following requests fail again, the consumer fails
with a PartitionNotFoundException.
If the producer has failed, the consumer is cancelled.
If the producer is not running and has not failed, there is a bug either in the
consumer task setup (e.g. requesting a non-existing result) or in the network
stack (e.g. unsafe publication of produced results), in which case the error is
attributed to the consumer.
---
The new Akka messages introduced with this change are only exchanged in error
cases and don't affect normal operation.
Normal operation (not affected by this change):
- TM1=>TM2: request result
- TM2=>TM1: result
Error case:
- TM1=>TM2: request result
- TM2=>TM1: PartitionNotFoundException
- TM1=>JM: check partition state
- JM=>TM1: retrigger request -OR- cancel consumer
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30821256
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -148,6 +148,16 @@
public static final String TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "taskmanager.network.defaultIOMode";
/**
+ * Initial backoff (in ms) for partition requests, which return partition not found.
+ */
+ public static final String TASK_MANAGER_NETWORK_REQUEST_BACKOFF_INITIAL_MS = "taskmanager.network.requestBackoffInitial";
--- End diff --
OK, I'll remove the configuration values and just have the default values hard coded. For the integration test, I can easily get the NetworkEnvironment and overwrite the values via Reflection.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/705#issuecomment-104335429
I think this looks good.
+1 to get it in fast and give it a bit of exposure before the release.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30817476
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -148,6 +148,16 @@
public static final String TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "taskmanager.network.defaultIOMode";
/**
+ * Initial backoff (in ms) for partition requests, which return partition not found.
+ */
+ public static final String TASK_MANAGER_NETWORK_REQUEST_BACKOFF_INITIAL_MS = "taskmanager.network.requestBackoffInitial";
--- End diff --
Actually no. I think it's safe to have it fixed. We only need to adjust the values for tests. What is the easiest way to only allow internal configuration? The problem is that for integration tests it's hard to set configuration values for runtime components otherwise.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30817061
--- Diff: flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java ---
@@ -30,20 +30,20 @@
import org.junit.Test;
public class ClassLoaderITCase {
-
+
private static final String INPUT_SPLITS_PROG_JAR_FILE = "target/customsplit-test-jar.jar";
private static final String STREAMING_PROG_JAR_FILE = "target/streamingclassloader-test-jar.jar";
private static final String KMEANS_JAR_PATH = "target/kmeans-test-jar.jar";
-
+
@Test
public void testJobWithCustomInputFormat() {
try {
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
-
--- End diff --
Whitespace reformat?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30817054
--- Diff: flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java ---
@@ -27,7 +27,6 @@
protected String textPath;
protected String resultPath;
-
--- End diff --
Whitespace reformat?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30821430
--- Diff: flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java ---
@@ -30,20 +30,20 @@
import org.junit.Test;
public class ClassLoaderITCase {
-
+
private static final String INPUT_SPLITS_PROG_JAR_FILE = "target/customsplit-test-jar.jar";
private static final String STREAMING_PROG_JAR_FILE = "target/streamingclassloader-test-jar.jar";
private static final String KMEANS_JAR_PATH = "target/kmeans-test-jar.jar";
-
+
@Test
public void testJobWithCustomInputFormat() {
try {
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
-
--- End diff --
I know. I couldn't find the option to disable this in Intellij (clears white-space only lines).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30821561
--- Diff: flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java ---
@@ -27,7 +27,6 @@
protected String textPath;
protected String resultPath;
-
--- End diff --
I know we should refrain from whitespace reformats. It was accidental, but I think it's OK (two empty lines). I agree that we shouldn't do whitespace reformats, but I don't think that these are affecting review-ability of this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/705#issuecomment-104643115
I've removed the configuration constants and Travis passed after the rebase (modulo KafkaITCase). I'm merging this...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/705
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30816631
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -148,6 +148,16 @@
public static final String TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "taskmanager.network.defaultIOMode";
/**
+ * Initial backoff (in ms) for partition requests, which return partition not found.
+ */
+ public static final String TASK_MANAGER_NETWORK_REQUEST_BACKOFF_INITIAL_MS = "taskmanager.network.requestBackoffInitial";
--- End diff --
Are these config parameters that we want to expose? Or would it be safe to make those internal, if there really is no good case for anyone to ever adjust them?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1954] [FLINK-1636] [runtime] Improve pa...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/705#discussion_r30817964
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -148,6 +148,16 @@
public static final String TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "taskmanager.network.defaultIOMode";
/**
+ * Initial backoff (in ms) for partition requests, which return partition not found.
+ */
+ public static final String TASK_MANAGER_NETWORK_REQUEST_BACKOFF_INITIAL_MS = "taskmanager.network.requestBackoffInitial";
--- End diff --
For unit tests, you could try to pass them directly to the network environment.
For system integration tests, it is tricky. You may use some reflection magic modify the constants (sneaky, but test only).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---