flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-4024) FileSourceFunction not adjusted to new IF lifecycle
Date Tue, 14 Jun 2016 16:43:57 GMT

    [ https://issues.apache.org/jira/browse/FLINK-4024?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15329842#comment-15329842
] 

ASF GitHub Bot commented on FLINK-4024:
---------------------------------------

Github user kl0u commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2100#discussion_r67008351
  
    --- 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;
    --- End diff --
    
    Same for this one.


> 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)

Mime
View raw message