You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2016/06/16 07:40:42 UTC
[1/2] flink git commit: [FLINK-3908] [core] Fix resetting of
FieldParser error state.
Repository: flink
Updated Branches:
refs/heads/master 867be9b6a -> fe0eb602d
[FLINK-3908] [core] Fix resetting of FieldParser error state.
This closes #2007
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d8747978
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d8747978
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d8747978
Branch: refs/heads/master
Commit: d8747978386c0aacce1dc2631e1c5602ec9b2a00
Parents: 867be9b
Author: Flavio Pompermaier <f....@gmail.com>
Authored: Thu May 19 11:41:43 2016 +0200
Committer: Fabian Hueske <fh...@apache.org>
Committed: Wed Jun 15 23:43:26 2016 +0200
----------------------------------------------------------------------
.../api/common/io/GenericCsvInputFormat.java | 2 +-
.../flink/types/parser/ByteValueParser.java | 2 +-
.../apache/flink/types/parser/FieldParser.java | 28 +++++++++++++++++---
.../flink/types/parser/IntValueParser.java | 2 +-
.../apache/flink/types/parser/LongParser.java | 4 +--
.../flink/types/parser/LongValueParser.java | 4 +--
.../apache/flink/types/parser/ShortParser.java | 2 +-
.../flink/types/parser/ShortValueParser.java | 2 +-
.../apache/flink/types/parser/StringParser.java | 8 +++---
.../flink/types/parser/StringValueParser.java | 8 +++---
.../flink/api/java/io/PrimitiveInputFormat.java | 2 +-
11 files changed, 42 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
index 31d42ff..e2c54ad 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
@@ -392,7 +392,7 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
@SuppressWarnings("unchecked")
FieldParser<Object> parser = (FieldParser<Object>) this.fieldParsers[output];
Object reuse = holders[output];
- startPos = parser.parseField(bytes, startPos, limit, this.fieldDelim, reuse);
+ startPos = parser.resetErrorStateAndParse(bytes, startPos, limit, this.fieldDelim, reuse);
holders[output] = parser.getLastResult();
// check parse result
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/ByteValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/ByteValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/ByteValueParser.java
index 416a498..4cda98c 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/ByteValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/ByteValueParser.java
@@ -38,7 +38,7 @@ public class ByteValueParser extends FieldParser<ByteValue> {
this.result = reusable;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
if (bytes[startPos] == '-') {
neg = true;
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/FieldParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/FieldParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/FieldParser.java
index 57689a8..67c1bd7 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/FieldParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/FieldParser.java
@@ -79,8 +79,9 @@ public abstract class FieldParser<T> {
private ParseErrorState errorState = ParseErrorState.NONE;
/**
- * Parses the value of a field from the byte array.
- * The start position within the byte array and the array's valid length is given.
+ * Parses the value of a field from the byte array, taking care of properly reset
+ * the state of this parser.
+ * The start position within the byte array and the array's valid length is given.
* The content of the value is delimited by a field delimiter.
*
* @param bytes The byte array that holds the value.
@@ -92,8 +93,27 @@ public abstract class FieldParser<T> {
*
* @return The index of the next delimiter, if the field was parsed correctly. A value less than 0 otherwise.
*/
- public abstract int parseField(byte[] bytes, int startPos, int limit, byte[] delim, T reuse);
-
+ public int resetErrorStateAndParse(byte[] bytes, int startPos, int limit, byte[] delim, T reuse) {
+ resetParserState();
+ return parseField(bytes, startPos, limit, delim, reuse);
+ }
+
+ /**
+ * Each parser's logic should be implemented inside this method
+ *
+ * @see {@link FieldParser#parseField(byte[], int, int, byte[], Object)}
+ * */
+ protected abstract int parseField(byte[] bytes, int startPos, int limit, byte[] delim, T reuse);
+
+ /**
+ * Reset the state of the parser. Called as the very first method inside
+ * {@link FieldParser#resetErrorStateAndParse(byte[], int, int, byte[], Object)}, by default it just reset
+ * its error state.
+ * */
+ protected void resetParserState() {
+ this.errorState = ParseErrorState.NONE;
+ }
+
/**
* Gets the parsed field. This method returns the value parsed by the last successful invocation of
* {@link #parseField(byte[], int, int, byte[], Object)}. It objects are mutable and reused, it will return
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/IntValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/IntValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/IntValueParser.java
index d493e2d..d487c66 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/IntValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/IntValueParser.java
@@ -39,7 +39,7 @@ public class IntValueParser extends FieldParser<IntValue> {
long val = 0;
boolean neg = false;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
this.result = reusable;
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/LongParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/LongParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/LongParser.java
index e8ac09b..c7b76d2 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/LongParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/LongParser.java
@@ -72,7 +72,7 @@ public class LongParser extends FieldParser<Long> {
if (i+1 >= limit) {
return limit;
- } else if (i+1 < delimLimit && delimiterNext(bytes, i+1, delimiter)) {
+ } else if (i + 1 < delimLimit && delimiterNext(bytes, i + 1, delimiter)) {
return i + 1 + delimiter.length;
} else {
setErrorState(ParseErrorState.NUMERIC_VALUE_OVERFLOW_UNDERFLOW);
@@ -160,7 +160,7 @@ public class LongParser extends FieldParser<Long> {
if (val < 0) {
// this is an overflow/underflow, unless we hit exactly the Long.MIN_VALUE
if (neg && val == Long.MIN_VALUE) {
- if (length == 1 || bytes[startPos+1] == delimiter) {
+ if (length == 1 || bytes[startPos + 1] == delimiter) {
return Long.MIN_VALUE;
} else {
throw new NumberFormatException("value overflow");
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/LongValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/LongValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/LongValueParser.java
index eaaf619..597abc0 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/LongValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/LongValueParser.java
@@ -36,7 +36,7 @@ public class LongValueParser extends FieldParser<LongValue> {
long val = 0;
boolean neg = false;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
this.result = reusable;
@@ -75,7 +75,7 @@ public class LongValueParser extends FieldParser<LongValue> {
if (i+1 >= limit) {
return limit;
- } else if (i+1 < delimLimit && delimiterNext(bytes, i+1, delimiter)) {
+ } else if (i + 1 < delimLimit && delimiterNext(bytes, i + 1, delimiter)) {
return i + 1 + delimiter.length;
} else {
setErrorState(ParseErrorState.NUMERIC_VALUE_OVERFLOW_UNDERFLOW);
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/ShortParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/ShortParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/ShortParser.java
index 6f1fc7b..3afa761 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/ShortParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/ShortParser.java
@@ -39,7 +39,7 @@ public class ShortParser extends FieldParser<Short> {
int val = 0;
boolean neg = false;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
if (bytes[startPos] == '-') {
neg = true;
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/ShortValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/ShortValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/ShortValueParser.java
index c1df9ce..880af25 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/ShortValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/ShortValueParser.java
@@ -39,7 +39,7 @@ public class ShortValueParser extends FieldParser<ShortValue> {
int val = 0;
boolean neg = false;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
this.result = reusable;
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
index 7dc80e5..9cee990 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
@@ -44,14 +44,14 @@ public class StringParser extends FieldParser<String> {
int i = startPos;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
if(quotedStringParsing && bytes[i] == quoteCharacter) {
// quoted string parsing enabled and first character Vis a quote
i++;
// search for ending quote character, continue when it is escaped
- while (i < limit && (bytes[i] != quoteCharacter || bytes[i-1] == BACKSLASH)){
+ while (i < limit && (bytes[i] != quoteCharacter || bytes[i - 1] == BACKSLASH)) {
i++;
}
@@ -63,11 +63,11 @@ public class StringParser extends FieldParser<String> {
// check for proper termination
if (i == limit) {
// either by end of line
- this.result = new String(bytes, startPos+1, i - startPos - 2);
+ this.result = new String(bytes, startPos + 1, i - startPos - 2);
return limit;
} else if ( i < delimLimit && delimiterNext(bytes, i, delimiter)) {
// or following field delimiter
- this.result = new String(bytes, startPos+1, i - startPos - 2);
+ this.result = new String(bytes, startPos + 1, i - startPos - 2);
return i + delimiter.length;
} else {
// no proper termination
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
index 586c2cc..bc89952 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
@@ -48,14 +48,14 @@ public class StringValueParser extends FieldParser<StringValue> {
this.result = reusable;
int i = startPos;
- final int delimLimit = limit-delimiter.length+1;
+ final int delimLimit = limit - delimiter.length + 1;
if(quotedStringParsing == true && bytes[i] == quoteCharacter) {
// quoted string parsing enabled and first character is a quote
i++;
// search for ending quote character, continue when it is escaped
- while (i < limit && (bytes[i] != quoteCharacter || bytes[i-1] == BACKSLASH)){
+ while (i < limit && (bytes[i] != quoteCharacter || bytes[i - 1] == BACKSLASH)) {
i++;
}
@@ -67,11 +67,11 @@ public class StringValueParser extends FieldParser<StringValue> {
// check for proper termination
if (i == limit) {
// either by end of line
- reusable.setValueAscii(bytes, startPos+1, i - startPos - 2);
+ reusable.setValueAscii(bytes, startPos + 1, i - startPos - 2);
return limit;
} else if ( i < delimLimit && delimiterNext(bytes, i, delimiter)) {
// or following field delimiter
- reusable.setValueAscii(bytes, startPos+1, i - startPos - 2);
+ reusable.setValueAscii(bytes, startPos + 1, i - startPos - 2);
return i + delimiter.length;
} else {
// no proper termination
http://git-wip-us.apache.org/repos/asf/flink/blob/d8747978/flink-java/src/main/java/org/apache/flink/api/java/io/PrimitiveInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/PrimitiveInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/PrimitiveInputFormat.java
index 3dbc966..75b82cd 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/PrimitiveInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/PrimitiveInputFormat.java
@@ -75,7 +75,7 @@ public class PrimitiveInputFormat<OT> extends DelimitedInputFormat<OT> {
}
// Null character as delimiter is used because there's only 1 field to be parsed
- if (parser.parseField(bytes, offset, numBytes + offset, new byte[]{'\0'}, reuse) >= 0) {
+ if (parser.resetErrorStateAndParse(bytes, offset, numBytes + offset, new byte[]{'\0'}, reuse) >= 0) {
return parser.getLastResult();
} else {
String s = new String(bytes, offset, numBytes);
[2/2] flink git commit: [FLINK-4024] [streaming] Fix InputFormat
lifecycle in FileSourceFunction.
Posted by fh...@apache.org.
[FLINK-4024] [streaming] Fix InputFormat lifecycle in FileSourceFunction.
This closes #2100
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fe0eb602
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fe0eb602
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fe0eb602
Branch: refs/heads/master
Commit: fe0eb602d2e908d116d71018bb141e5c748c780b
Parents: d874797
Author: kl0u <kk...@gmail.com>
Authored: Mon Jun 13 11:55:56 2016 +0200
Committer: Fabian Hueske <fh...@apache.org>
Committed: Wed Jun 15 23:43:31 2016 +0200
----------------------------------------------------------------------
.../environment/StreamExecutionEnvironment.java | 4 +-
.../source/ContinuousFileReaderOperator.java | 8 +
.../api/functions/source/InputFormatSource.java | 148 ----------
.../source/InputFormatSourceFunction.java | 170 +++++++++++
.../api/graph/StreamGraphGenerator.java | 6 +-
.../source/InputFormatSourceFunctionTest.java | 286 +++++++++++++++++++
6 files changed, 469 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index d6bfc02..a2a38d4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -56,7 +56,7 @@ import org.apache.flink.streaming.api.functions.source.FilePathFilter;
import org.apache.flink.streaming.api.functions.source.FileReadFunction;
import org.apache.flink.streaming.api.functions.source.ContinuousFileMonitoringFunction;
import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
-import org.apache.flink.streaming.api.functions.source.InputFormatSource;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
import org.apache.flink.streaming.api.functions.source.FromElementsFunction;
import org.apache.flink.streaming.api.functions.source.FromIteratorFunction;
import org.apache.flink.streaming.api.functions.source.FromSplittableIteratorFunction;
@@ -1215,7 +1215,7 @@ public abstract class StreamExecutionEnvironment {
TypeInformation<OUT> typeInfo,
String sourceName) {
- InputFormatSource<OUT> function = new InputFormatSource<>(inputFormat, typeInfo);
+ InputFormatSourceFunction<OUT> function = new InputFormatSourceFunction<>(inputFormat, typeInfo);
return addSource(function, sourceName, typeInfo);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
index 4d4a792..e26c534 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
@@ -97,7 +97,9 @@ public class ContinuousFileReaderOperator<OUT, S extends Serializable> extends A
"Please report it.");
}
+ this.format.setRuntimeContext(getRuntimeContext());
this.format.configure(new Configuration());
+
this.collector = new TimestampedCollector<>(output);
this.checkpointLock = getContainingTask().getCheckpointLock();
@@ -230,9 +232,13 @@ public class ContinuousFileReaderOperator<OUT, S extends Serializable> extends A
@Override
public void run() {
try {
+
+ this.format.openInputFormat();
+
while (this.isRunning) {
synchronized (checkpointLock) {
+
if (this.currentSplit != null) {
if (currentSplit.equals(EOS)) {
@@ -297,6 +303,8 @@ public class ContinuousFileReaderOperator<OUT, S extends Serializable> extends A
} finally {
synchronized (checkpointLock) {
LOG.info("Reader terminated, and exiting...");
+ this.format.closeInputFormat();
+ this.isRunning = false;
checkpointLock.notifyAll();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSource.java
deleted file mode 100644
index 2a84781..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSource.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.streaming.api.functions.source;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.io.InputFormat;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.io.InputSplit;
-import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
-import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-@Internal
-public class InputFormatSource<OUT> extends RichParallelSourceFunction<OUT> {
- private static final long serialVersionUID = 1L;
-
- private TypeInformation<OUT> typeInfo;
- private transient TypeSerializer<OUT> serializer;
-
- private InputFormat<OUT, InputSplit> format;
-
- private transient InputSplitProvider provider;
- private transient Iterator<InputSplit> splitIterator;
-
- private volatile boolean isRunning = true;
-
- @SuppressWarnings("unchecked")
- public InputFormatSource(InputFormat<OUT, ?> format, TypeInformation<OUT> typeInfo) {
- this.format = (InputFormat<OUT, InputSplit>) format;
- this.typeInfo = typeInfo;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public void open(Configuration parameters) throws Exception {
- StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
- this.provider = context.getInputSplitProvider();
-
- format.configure(parameters);
- serializer = typeInfo.createSerializer(getRuntimeContext().getExecutionConfig());
-
- splitIterator = getInputSplits();
- if (splitIterator.hasNext()) {
- format.open(splitIterator.next());
- }
- isRunning = true;
- }
-
- @Override
- public void close() throws Exception {
- format.close();
- }
-
- private Iterator<InputSplit> getInputSplits() {
-
- return new Iterator<InputSplit>() {
-
- private InputSplit nextSplit;
-
- private boolean exhausted;
-
- @Override
- public boolean hasNext() {
- if (exhausted) {
- return false;
- }
-
- if (nextSplit != null) {
- return true;
- }
-
- InputSplit split = provider.getNextInputSplit();
-
- if (split != null) {
- this.nextSplit = split;
- return true;
- } else {
- exhausted = true;
- return false;
- }
- }
-
- @Override
- public InputSplit next() {
- if (this.nextSplit == null && !hasNext()) {
- throw new NoSuchElementException();
- }
-
- final InputSplit tmp = this.nextSplit;
- this.nextSplit = null;
- return tmp;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
-
- @Override
- public void run(SourceContext<OUT> ctx) throws Exception {
- while (isRunning) {
- OUT nextElement = serializer.createInstance();
- nextElement = format.nextRecord(nextElement);
- if (nextElement == null && splitIterator.hasNext()) {
- format.open(splitIterator.next());
- continue;
- } else if (nextElement == null) {
- break;
- }
- ctx.collect(nextElement);
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
-
- /**
- * Returns the {@code InputFormat}. This is only needed because we need to set the input
- * split assigner on the {@code StreamGraph}.
- */
- public InputFormat<OUT, InputSplit> getFormat() {
- return format;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
new file mode 100644
index 0000000..bce1ec5
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
@@ -0,0 +1,170 @@
+/*
+ * 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.streaming.api.functions.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.io.RichInputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+@Internal
+public class InputFormatSourceFunction<OUT> extends RichParallelSourceFunction<OUT> {
+ private static final long serialVersionUID = 1L;
+
+ private TypeInformation<OUT> typeInfo;
+ private transient TypeSerializer<OUT> serializer;
+
+ private InputFormat<OUT, InputSplit> format;
+
+ private transient InputSplitProvider provider;
+ private transient Iterator<InputSplit> splitIterator;
+
+ private volatile boolean isRunning = true;
+
+ @SuppressWarnings("unchecked")
+ public InputFormatSourceFunction(InputFormat<OUT, ?> format, TypeInformation<OUT> typeInfo) {
+ this.format = (InputFormat<OUT, InputSplit>) format;
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void open(Configuration parameters) throws Exception {
+ StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
+
+ if (format instanceof RichInputFormat) {
+ ((RichInputFormat) format).setRuntimeContext(context);
+ }
+ format.configure(parameters);
+
+ provider = context.getInputSplitProvider();
+ serializer = typeInfo.createSerializer(getRuntimeContext().getExecutionConfig());
+ splitIterator = getInputSplits();
+ isRunning = splitIterator.hasNext();
+ }
+
+ @Override
+ public void run(SourceContext<OUT> ctx) throws Exception {
+ try {
+
+ if (isRunning && format instanceof RichInputFormat) {
+ ((RichInputFormat) format).openInputFormat();
+ }
+
+ OUT nextElement = serializer.createInstance();
+ while (isRunning) {
+ format.open(splitIterator.next());
+
+ // for each element we also check if cancel
+ // was called by checking the isRunning flag
+
+ while (isRunning && !format.reachedEnd()) {
+ nextElement = format.nextRecord(nextElement);
+ ctx.collect(nextElement);
+ }
+ format.close();
+
+ if (isRunning) {
+ isRunning = splitIterator.hasNext();
+ }
+ }
+ } finally {
+ format.close();
+ if (format instanceof RichInputFormat) {
+ ((RichInputFormat) format).closeInputFormat();
+ }
+ isRunning = false;
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isRunning = false;
+ }
+
+ @Override
+ public void close() throws Exception {
+ format.close();
+ if (format instanceof RichInputFormat) {
+ ((RichInputFormat) format).closeInputFormat();
+ }
+ }
+
+ /**
+ * Returns the {@code InputFormat}. This is only needed because we need to set the input
+ * split assigner on the {@code StreamGraph}.
+ */
+ public InputFormat<OUT, InputSplit> getFormat() {
+ return format;
+ }
+
+ private Iterator<InputSplit> getInputSplits() {
+
+ return new Iterator<InputSplit>() {
+
+ private InputSplit nextSplit;
+
+ private boolean exhausted;
+
+ @Override
+ public boolean hasNext() {
+ if (exhausted) {
+ return false;
+ }
+
+ if (nextSplit != null) {
+ return true;
+ }
+
+ InputSplit split = provider.getNextInputSplit();
+
+ if (split != null) {
+ this.nextSplit = split;
+ return true;
+ } else {
+ exhausted = true;
+ return false;
+ }
+ }
+
+ @Override
+ public InputSplit next() {
+ if (this.nextSplit == null && !hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ final InputSplit tmp = this.nextSplit;
+ this.nextSplit = null;
+ return tmp;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index 685655e..de80e25 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -21,7 +21,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.InputFormatSource;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
import org.apache.flink.streaming.api.transformations.CoFeedbackTransformation;
import org.apache.flink.streaming.api.transformations.FeedbackTransformation;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
@@ -425,8 +425,8 @@ public class StreamGraphGenerator {
null,
source.getOutputType(),
"Source: " + source.getName());
- if (source.getOperator().getUserFunction() instanceof InputFormatSource) {
- InputFormatSource<T> fs = (InputFormatSource<T>) source.getOperator().getUserFunction();
+ if (source.getOperator().getUserFunction() instanceof InputFormatSourceFunction) {
+ InputFormatSourceFunction<T> fs = (InputFormatSourceFunction<T>) source.getOperator().getUserFunction();
streamGraph.setInputFormat(source.getId(), fs.getFormat());
}
streamGraph.setParallelism(source.getId(), source.getParallelism());
http://git-wip-us.apache.org/repos/asf/flink/blob/fe0eb602/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunctionTest.java
new file mode 100644
index 0000000..32776e1
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunctionTest.java
@@ -0,0 +1,286 @@
+/*
+ * 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.streaming.api.functions.source;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.io.RichInputFormat;
+import org.apache.flink.api.common.io.statistics.BaseStatistics;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.core.io.InputSplitAssigner;
+import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+public class InputFormatSourceFunctionTest {
+
+ @Test
+ public void testNormalOp() throws Exception {
+ testFormatLifecycle(false);
+ }
+
+ @Test
+ public void testCancelation() throws Exception {
+ testFormatLifecycle(true);
+ }
+
+ private void testFormatLifecycle(final boolean midCancel) throws Exception {
+
+ final int noOfSplits = 5;
+ final int cancelAt = 2;
+
+ final LifeCycleTestInputFormat format = new LifeCycleTestInputFormat();
+ final InputFormatSourceFunction<Integer> reader = new InputFormatSourceFunction<>(format, TypeInformation.of(Integer.class));
+ reader.setRuntimeContext(new MockRuntimeContext(format, noOfSplits));
+
+ Assert.assertTrue(!format.isConfigured);
+ Assert.assertTrue(!format.isInputFormatOpen);
+ Assert.assertTrue(!format.isSplitOpen);
+
+ reader.open(new Configuration());
+ Assert.assertTrue(format.isConfigured);
+
+ TestSourceContext ctx = new TestSourceContext(reader, format, midCancel, cancelAt);
+ reader.run(ctx);
+
+ int splitsSeen = ctx.getSplitsSeen();
+ Assert.assertTrue(midCancel ? splitsSeen == cancelAt : splitsSeen == noOfSplits);
+
+ // we have exhausted the splits so the
+ // format and splits should be closed by now
+
+ Assert.assertTrue(!format.isSplitOpen);
+ Assert.assertTrue(!format.isInputFormatOpen);
+ }
+
+
+ private static class LifeCycleTestInputFormat extends RichInputFormat<Integer,InputSplit> {
+
+ private boolean isConfigured = false;
+ private boolean isInputFormatOpen = false;
+ private boolean isSplitOpen = false;
+
+ // end of split
+ private boolean eos = false;
+
+ private int splitCounter = 0;
+
+ private int reachedEndCalls = 0;
+ private int nextRecordCalls = 0;
+
+ @Override
+ public void openInputFormat() {
+ Assert.assertTrue(isConfigured);
+ Assert.assertTrue(!isInputFormatOpen);
+ Assert.assertTrue(!isSplitOpen);
+ this.isInputFormatOpen = true;
+ }
+
+ @Override
+ public void closeInputFormat() {
+ Assert.assertTrue(!isSplitOpen);
+ this.isInputFormatOpen = false;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {
+ Assert.assertTrue(!isConfigured);
+ this.isConfigured = true;
+ }
+
+ @Override
+ public BaseStatistics getStatistics(BaseStatistics cachedStatistics) throws IOException {
+ return null;
+ }
+
+ @Override
+ public InputSplit[] createInputSplits(int minNumSplits) throws IOException {
+ Assert.assertTrue(isConfigured);
+ InputSplit[] splits = new InputSplit[minNumSplits];
+ for (int i = 0; i < minNumSplits; i++) {
+ final int idx = i;
+ splits[idx] = new InputSplit() {
+ @Override
+ public int getSplitNumber() {
+ return idx;
+ }
+ };
+ }
+ return splits;
+ }
+
+ @Override
+ public InputSplitAssigner getInputSplitAssigner(InputSplit[] inputSplits) {
+ return null;
+ }
+
+ @Override
+ public void open(InputSplit split) throws IOException {
+ // whenever a new split opens,
+ // the previous should have been closed
+ Assert.assertTrue(isInputFormatOpen);
+ Assert.assertTrue(isConfigured);
+ Assert.assertTrue(!isSplitOpen);
+
+ isSplitOpen = true;
+ eos = false;
+ }
+
+ @Override
+ public boolean reachedEnd() throws IOException {
+ Assert.assertTrue(isInputFormatOpen);
+ Assert.assertTrue(isConfigured);
+ Assert.assertTrue(isSplitOpen);
+
+ if (!eos) {
+ reachedEndCalls++;
+ }
+ return eos;
+ }
+
+ @Override
+ public Integer nextRecord(Integer reuse) throws IOException {
+ Assert.assertTrue(isInputFormatOpen);
+ Assert.assertTrue(isConfigured);
+ Assert.assertTrue(isSplitOpen);
+
+ Assert.assertTrue(reachedEndCalls == ++nextRecordCalls);
+
+ eos = true;
+ return splitCounter++;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.isSplitOpen = false;
+ }
+ }
+
+ private static class TestSourceContext implements SourceFunction.SourceContext<Integer> {
+
+ private final InputFormatSourceFunction<Integer> reader;
+ private final LifeCycleTestInputFormat format;
+ private final boolean shouldCancel;
+ private final int cancelAt;
+
+ int splitIdx = 0;
+
+ private TestSourceContext(InputFormatSourceFunction<Integer> reader, LifeCycleTestInputFormat format, boolean shouldCancel, int cancelAt) {
+ this.reader = reader;
+ this.format = format;
+ this.shouldCancel = shouldCancel;
+ this.cancelAt = cancelAt;
+ }
+
+ @Override
+ public void collect(Integer element) {
+ Assert.assertTrue(format.isSplitOpen);
+ Assert.assertTrue(splitIdx == element);
+ if (shouldCancel && splitIdx == cancelAt) {
+ reader.cancel();
+ } else {
+ splitIdx++;
+ }
+ }
+
+ @Override
+ public void collectWithTimestamp(Integer element, long timestamp) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ throw new UnsupportedOperationException();
+ }
+
+ public int getSplitsSeen() {
+ return this.splitIdx;
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private static class MockRuntimeContext extends StreamingRuntimeContext {
+
+ private final int noOfSplits;
+ private int nextSplit = 0;
+ private final LifeCycleTestInputFormat format;
+ private InputSplit[] inputSplits;
+
+ private MockRuntimeContext(LifeCycleTestInputFormat format, int noOfSplits) {
+
+ super(new MockStreamOperator(),
+ new MockEnvironment("no", 4 * MemoryManager.DEFAULT_PAGE_SIZE, null, 16),
+ Collections.<String, Accumulator<?, ?>>emptyMap());
+
+ this.noOfSplits = noOfSplits;
+ this.format = format;
+ }
+
+ @Override
+ public InputSplitProvider getInputSplitProvider() {
+ try {
+ this.inputSplits = format.createInputSplits(noOfSplits);
+ Assert.assertTrue(inputSplits.length == noOfSplits);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+
+ return new InputSplitProvider() {
+ @Override
+ public InputSplit getNextInputSplit() {
+ if (nextSplit < inputSplits.length) {
+ return inputSplits[nextSplit++];
+ }
+ return null;
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class MockStreamOperator extends AbstractStreamOperator<Integer> {
+ private static final long serialVersionUID = -1153976702711944427L;
+
+ @Override
+ public ExecutionConfig getExecutionConfig() {
+ return new ExecutionConfig();
+ }
+ }
+ }
+}