Mailing-List: contact dev-help@flink.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.incubator.apache.org
Date: Thu, 10 Jul 2014 13:34:05 +0000 (UTC)
From: "Gyula Fora (JIRA)" <jira@apache.org>
To: dev@flink.incubator.apache.org
Message-ID: <JIRA.12726586.1404999190296.15723.1404999245122@arcas>
In-Reply-To: <JIRA.12726586.1404999190296@arcas>
References: <JIRA.12726586.1404999190296@arcas>
Subject: [jira] [Created] (FLINK-1021) IllegalStateException at InputGate
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit

Gyula Fora created FLINK-1021:
---------------------------------

             Summary: IllegalStateException at InputGate
                 Key: FLINK-1021
                 URL: https://issues.apache.org/jira/browse/FLINK-1021
             Project: Flink
          Issue Type: Bug
          Components: Local Runtime
    Affects Versions: 0.6-incubating
         Environment: Linux/Windows

            Reporter: Gyula Fora


Until now, after every emit to the outputs we flushed them using the .flush() method of the recordwriter. Now we removed this flush() call and we have two interesting observations:

First of all we dont send enough records the source finishes but the output buffer never gets flushed.

Secondly if we generate a simple datastream from lets say the first 1500 numbers we get an exception in the InputGates (after lets say a hundred records): java.lang.IllegalStateException: Channel received an event before completing the current partial record.

java.lang.IllegalStateException: Channel received an event before completing the current partial record.
	at eu.stratosphere.runtime.io.channels.InputChannel.readRecord(InputChannel.java:177)
	at eu.stratosphere.runtime.io.gates.InputGate.readRecord(InputGate.java:173)
	at eu.stratosphere.streaming.api.streamcomponent.StreamRecordReader.hasNext(StreamRecordReader.java:96)
	at eu.stratosphere.streaming.api.streamcomponent.AbstractStreamComponent.invokeRecords(AbstractStreamComponent.java:255)
	at eu.stratosphere.streaming.api.streamcomponent.StreamSink.invoke(StreamSink.java:74)
	at eu.stratosphere.nephele.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:260)
	at java.lang.Thread.run(Unknown Source)

To produce the error run this test: https://github.com/stratosphere/stratosphere-streaming/blob/output-flush/stratosphere-streaming-core/src/test/java/eu/stratosphere/streaming/api/PrintTest.java

Please note that this is the output-flush branch in Stratoshpere-streaming

This works perfectly if we flush the outputs after the emits.


--
This message was sent by Atlassian JIRA
(v6.2#6252)