[
https://issues.apache.org/jira/browse/FLINK-4024?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15329784#comment-15329784
]
ASF GitHub Bot commented on FLINK-4024:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/2100#discussion_r67003727
--- Diff:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/FileSourceFunctionTest.java
---
@@ -0,0 +1,258 @@
+/*
+ * 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 FileSourceFunctionTest {
+
+ @Test
+ public void testNormalOp() throws Exception {
+ testForEmptyLocation(false);
+ }
+
+ @Test
+ public void testCancelation() throws Exception {
+ testForEmptyLocation(true);
+ }
+
+ private void testForEmptyLocation(final boolean midCancel) throws
Exception {
+
+ final int noOfSplits = 5;
+ final int cancelAt = 2;
+
+ final LifeCycleTestInputFormat format = new
LifeCycleTestInputFormat();
+ final FileSourceFunction<Integer> reader = new
FileSourceFunction<>(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;
+
+ @Override
+ public void openInputFormat() {
+ this.isInputFormatOpen = true;
+ }
+
+ @Override
+ public void closeInputFormat() {
+ this.isInputFormatOpen = false;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {
+ this.isConfigured = true;
+ }
+
+ @Override
+ public BaseStatistics getStatistics(BaseStatistics
cachedStatistics) throws IOException {
+ return null;
+ }
+
+ @Override
+ public InputSplit[] createInputSplits(int minNumSplits) throws
IOException {
+ 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(!isSplitOpen);
+
+ isSplitOpen = true;
+ eos = false;
+ }
+
+ @Override
+ public boolean reachedEnd() throws IOException {
+ return eos;
+ }
+
+ @Override
+ public Integer nextRecord(Integer reuse) throws IOException {
+ eos = true;
+ return splitCounter++;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.isSplitOpen = false;
--- End diff --
Assert that split was opened before?
> FileSourceFunction not adjusted to new IF lifecycle
> ---------------------------------------------------
>
> Key: FLINK-4024
> URL: https://issues.apache.org/jira/browse/FLINK-4024
> Project: Flink
> Issue Type: Bug
> Components: Streaming
> Affects Versions: 1.1.0
> Reporter: Chesnay Schepler
> Assignee: Kostas Kloudas
> Priority: Critical
> Fix For: 1.1.0
>
>
> The InputFormat lifecycle was extended in
> ac2137cfa5e63bd4f53a4b7669dc591ab210093f, adding additional
> open-/closeInputFormat() methods.
> The streaming FileSourceFunction was not adjusted for this change, and thus
> will fail for every InputFormat that leverages these new methods.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
