You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by fpompermaier <gi...@git.apache.org> on 2016/04/18 13:47:22 UTC
[GitHub] flink pull request: Managed closeInputFormat()
GitHub user fpompermaier opened a pull request:
https://github.com/apache/flink/pull/1903
Managed closeInputFormat()
Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful description of your changes.
- [X] General
- The pull request references the related JIRA issue
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message
- [X] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [X] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis build has passed
--------------
Added Javadocs, don't know wheter other documentation needs to be updated
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/fpompermaier/flink FLINK-3777
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1903.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 #1903
----
commit bbe0d8f9cf04a533da5642969270cc382b6c1d5b
Author: Flavio Pompermaier <f....@gmail.com>
Date: 2016-04-18T11:44:49Z
Managed closeInputFormat()
----
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-214734647
Merging.
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60727958
--- Diff: flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java ---
@@ -150,6 +151,12 @@ protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics
}
return;
}
+ finally {
+ // close the instantiated resources before returning
+ if(format != null && format instanceof RichInputFormat) {
--- End diff --
If we close the IF here, we might not be able to retrieve the stats.
---
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-3777] Managed closeInputFormat()
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/1903
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61064281
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java ---
@@ -84,16 +85,25 @@ public void testDataSourceTask() {
super.addOutput(this.outList);
DataSourceTask<Record> testTask = new DataSourceTask<>();
-
super.registerFileInputTask(testTask, MockInputFormat.class, new File(tempTestPath).toURI().toString(), "\n");
-
+
try {
testTask.invoke();
} catch (Exception e) {
System.err.println(e);
Assert.fail("Invoke method caused exception.");
}
+ try {
+ Field formatField = DataSourceTask.class.getDeclaredField("format");
+ formatField.setAccessible(true);
+ MockInputFormat inputFormat = (MockInputFormat) formatField.get(testTask);
+ Assert.assertTrue("Invalid status of the input format. Expected for closed: true, Actual: "+inputFormat.opened, inputFormat.closed);
--- End diff --
missing spaces around +
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61065029
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/io/ReplicatingInputFormat.java ---
@@ -133,4 +133,18 @@ public RuntimeContext getRuntimeContext(){
throw new RuntimeException("The underlying input format to this ReplicatingInputFormat isn't context aware");
}
}
+
+ @Override
+ public void openInputFormat() {
+ if(this.replicatedIF instanceof RichInputFormat) {
+ ((RichInputFormat)this.replicatedIF).openInputFormat();
+ }
+ }
+
+ @Override
+ public void closeInputFormat() {
+ if(this.replicatedIF instanceof RichInputFormat) {
--- End diff --
missing space after if
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-214731704
Thanks @fpompermaier.
Good to merge
---
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: Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-211347732
there are a few formatting issues with missing spaces after if/braces.
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-213399764
Hi @fpompermaier, I thought about this again. I don't think it is a good idea to tie `closeInputFormat()` to `configure()` because `configure()` is called at different places such as the JobManager and in the optimizer to retrieve statistics. Hence, it is not a good idea to open resources in `configure()`.
We should rather add a `openInputFormat()` method to `RichInputFormat()` which is called directly before `open()`. Would you add such a method and remove the unnecessary `closeInputFormat()` calls, and tests that check that `openInputFormat()` and `closeInputFormat()` are correctly called in the life cycle of a RichInputFormat?
Sorry for not thinking about this earlier. Thanks, Fabian
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-213573556
Thanks for the quick update @fpompermaier! The changes look mostly good.
What is missing are tests that verify that `DataSourceTask` and `GenericDataSourceBase` correctly call the new methods as part of their IF life cycle. Can you extend `GenericDataSourceBaseTest` and `DataSourceTask` and add tests to validate that the methods are called (maybe by using a mocked `RichInputFormat`)? Thanks
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-214729047
Hi @fpompermaier, thanks for the quick update. I have one last comment.
The PR is good to merge once that is resolved. Thanks
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60794865
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/io/ReplicatingInputFormat.java ---
@@ -133,4 +133,18 @@ public RuntimeContext getRuntimeContext(){
throw new RuntimeException("The underlying input format to this ReplicatingInputFormat isn't context aware");
}
}
+
+ @Override
+ public void openInputFormat() {
+ if(this.replicatedIF instanceof RichInputFormat){
--- End diff --
Can you add a space before and after the parentheses of the condition, i.e., `if (condition) {`?
Please check the other changes as well.
---
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-3777] Managed closeInputFormat()
Posted by fpompermaier <gi...@git.apache.org>.
Github user fpompermaier commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60828060
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java ---
@@ -82,9 +82,6 @@ public void initializeOnMaster(ClassLoader loader) throws Exception {
catch (Throwable t) {
throw new Exception("Configuring the InputFormat (" + formatDescription + ") failed: " + t.getMessage(), t);
}
- finally {
--- End diff --
no..sorry :)
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60794593
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java ---
@@ -82,9 +82,6 @@ public void initializeOnMaster(ClassLoader loader) throws Exception {
catch (Throwable t) {
throw new Exception("Configuring the InputFormat (" + formatDescription + ") failed: " + t.getMessage(), t);
}
- finally {
--- End diff --
Is this change intended?
---
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-3777] Managed closeInputFormat()
Posted by fpompermaier <gi...@git.apache.org>.
Github user fpompermaier commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-213406637
Me and Stefano agree on this :)
I'll will edit the PR and commit today
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61064993
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java ---
@@ -210,10 +210,12 @@ public void accept(Visitor<Operator<?>> visitor) {
protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
@SuppressWarnings("unchecked")
InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
+ //open the input format
--- End diff --
this comment seems out-dated with the differentiation of configure() and openInputFormat()
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61078670
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java ---
@@ -84,17 +84,26 @@ public void testDataSourceTask() {
super.addOutput(this.outList);
DataSourceTask<Record> testTask = new DataSourceTask<>();
-
super.registerFileInputTask(testTask, MockInputFormat.class, new File(tempTestPath).toURI().toString(), "\n");
-
+
try {
testTask.invoke();
} catch (Exception e) {
System.err.println(e);
Assert.fail("Invoke method caused exception.");
}
- Assert.assertTrue("Invalid output size. Expected: "+(keyCnt*valCnt)+" Actual: "+this.outList.size(),
+ try {
+ Field formatField = DataSourceTask.class.getDeclaredField("format");
+ formatField.setAccessible(true);
+ MockInputFormat inputFormat = (MockInputFormat) formatField.get(testTask);
+ Assert.assertTrue("Invalid status of the input format. Expected for closed: true, Actual: " + inputFormat.closed, inputFormat.closed);
--- End diff --
The check for `inputFormat.opened` is missing, right?
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61064423
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java ---
@@ -84,16 +85,25 @@ public void testDataSourceTask() {
super.addOutput(this.outList);
DataSourceTask<Record> testTask = new DataSourceTask<>();
-
super.registerFileInputTask(testTask, MockInputFormat.class, new File(tempTestPath).toURI().toString(), "\n");
-
+
try {
testTask.invoke();
} catch (Exception e) {
System.err.println(e);
Assert.fail("Invoke method caused exception.");
}
+ try {
+ Field formatField = DataSourceTask.class.getDeclaredField("format");
+ formatField.setAccessible(true);
+ MockInputFormat inputFormat = (MockInputFormat) formatField.get(testTask);
+ Assert.assertTrue("Invalid status of the input format. Expected for closed: true, Actual: "+inputFormat.opened, inputFormat.closed);
--- End diff --
you're also printing ```if.opened``` as the actual value but check ```if.closed```.
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61064772
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java ---
@@ -192,6 +194,13 @@ else if (!this.taskCanceled) {
}
} finally {
BatchTask.clearWriters(eventualOutputs);
+ // --------------------------------------------------------------------
+ // Closing
+ // --------------------------------------------------------------------
+ if(this.format!=null && RichInputFormat.class.isAssignableFrom(this.format.getClass())) {
--- End diff --
missing space after if and around !=
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61064734
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java ---
@@ -200,6 +209,7 @@ else if (!this.taskCanceled) {
else {
LOG.debug(getLogString("Data source operator cancelled"));
}
+
--- End diff --
random new empty line
---
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-3777] Managed closeInputFormat()
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r61065095
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java ---
@@ -255,6 +268,18 @@ public Record readRecord(Record target, byte[] record, int offset, int numBytes)
target.setField(1, this.value);
return target;
}
+
+ public void openInputFormat() {
+ //ensure this is called only once
+ Assert.assertFalse("Invalid status of the input format. Expected for opened: false, Actual: "+opened, opened);
--- End diff --
missing spaces around + (and again in closeInputFormat
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60728010
--- Diff: flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java ---
@@ -187,6 +194,10 @@ else if (len >= 0) {
this.estimatedNumRecords = card;
}
}
+ // close the instantiated resources
+ if(format != null && format instanceof RichInputFormat) {
--- End diff --
IF should be closed after the stats have been retrieved (`format.getStatistics(cachedStatistics);`).
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-212483482
Thanks for this PR @fpompermaier.
We need to check all places where `InputFormat.configure()` is called to ensure that we free resources as promised in the JavaDocs of `RichInputFormat.closeInputFormat()`. I found a few more spots where we need to call `closeInputFormat()` if the IF is a `RichInputFormat`.
- ReplicatingInputFormat which is a wrapper for other input formats.
- DataSourceNode.computeOperatorSpecificDefaultEstimates() which tries to obtain some input statistics and initializes the input format with `configure()`.
- `InputFormatVertex.initializeOnMaster()` calls `configure()` as well.
It would also be good to add a test that the closeInputFormat() method is actually called.
---
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-3777] Managed closeInputFormat()
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1903#discussion_r60728363
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java ---
@@ -64,13 +65,19 @@ public void initializeOnMaster(ClassLoader loader) throws Exception {
}
// instantiate, if necessary
- InputFormat<?, ?> inputFormat;
+ InputFormat<?, ?> inputFormat = null;
try {
inputFormat = wrapper.getUserCodeObject(InputFormat.class, loader);
}
catch (Throwable t) {
throw new Exception("Instantiating the InputFormat (" + formatDescription + ") failed: " + t.getMessage(), t);
}
+ finally{
+ // close the instantiated resources before returning
+ if(inputFormat != null && inputFormat instanceof RichInputFormat) {
--- End diff --
`inputFormat.configure()` has not been called yet. Why are you closing it?
---
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-3777] Managed closeInputFormat()
Posted by fpompermaier <gi...@git.apache.org>.
Github user fpompermaier commented on the pull request:
https://github.com/apache/flink/pull/1903#issuecomment-214730690
I added the missing assert, now it should be ok!
---
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.
---