I am not sure about this... You are right about the super constructor, however, selectChannels(...) does not call super.getStrategy() what is the only way to get back the value set in the super class (ie, StreamPartitioner.strategy).
selectChannels() computes the return value independently from
this.forward and super.strategy...
-Matthias
On 06/10/2015 08:38 PM, Aljoscha Krettek wrote:
> It doesn't evaluate the member forward, but it calls the super constructor
> with a partitioning strategy that depends on on the forward parameter.
> That's how it works.
>
> On Wed, 10 Jun 2015 at 18:51 Márton Balassi <balassi.mar...@gmail.com>
> wrote:
>
>> Thanks for spotting the documentation issues. I'm fixing them quickly for
>> the release then.
>> The RebalancePartitioner indeed is obfuscated, let me see what can be done
>> there.
>>
>> On Wed, Jun 10, 2015 at 6:34 PM, Matthias J. Sax <
>> mj...@informatik.hu-berlin.de> wrote:
>>
>>> Thanks!
>>>
>>>
>>> About shuffle() vs rebalance(): I would suggest to explain the
>>> difference (random vs round-robin) in the JavaDoc of DataStream.
>>>
>>> Furthermore, I was wondering if the JavaDoc for @return is correct for
>>> "forward()", "rebalance()", and "global()". They all state
>>>
>>>> "@return The DataStream with shuffle partitioning set."
>>>
>>> (Looks like a copy&past error to me.)
>>>
>>> I am also wondering, if RebalancePartitioner has a bug. It seems, that
>>> it never evaluates its member "forward". Thus, local forward
>>> ("DataStream.forward()") would not work correctly.
>>>
>>> Please correct me, if I got something mixed up.
>>>
>>>
>>> -Matthias
>>>
>>> On 06/10/2015 02:42 PM, Márton Balassi wrote:
>>>> Hey,
>>>>
>>>> As the storm-compatibility-core build goes fine this is a dependency
>>> issue
>>>> with storm-compatibility-examples. As a first try replace:
>>>>
>>>> <dependency>
>>>> <groupId>org.apache.flink</groupId>
>>>> <artifactId>flink-streaming-core</artifactId>
>>>> <version>${project.version}</version>
>>>> <scope>test</scope>
>>>> <classifier>tests</classifier>
>>>> </dependency>
>>>>
>>>> with
>>>>
>>>> <dependency>
>>>> <groupId>org.apache.flink</groupId>
>>>> <artifactId>flink-streaming-core</artifactId>
>>>> <version>${project.version}</version>
>>>> <scope>test</scope>
>>>> <type>test-jar</type>
>>>> </dependency>
>>>>
>>>> And if you are already there please update the following as well for
>> the
>>>> renamings:
>>>>
>>>> <artifactId>flink-storm-examples</artifactId>
>>>> <name>flink-storm-examples</name>
>>>>
>>>> On Wed, Jun 10, 2015 at 2:22 PM, Matthias J. Sax <
>>>> mj...@informatik.hu-berlin.de> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> the current PR of storm compatibility layer builds successfully on my
>>>>> laptop (mvn clean install). However, on travis I get strange error
>>>>> messages in the IT-Cases:
>>>>> https://travis-ci.org/mjsax/flink/builds/66137928
>>>>>
>>>>> For example:
>>>>>
>>>>>> Caused by: java.lang.AbstractMethodError:
>>>>>
>>>
>> org.apache.flink.stormcompatibility.wrappers.StormFiniteSpoutWrapper.run(Lorg/apache/flink/streaming/api/functions/source/SourceFunction$SourceContext;)V
>>>>>> at
>>>>>
>>>
>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:49)
>>>>>> at
>>>>>
>>>
>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.invoke(SourceStreamTask.java:55)
>>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>>>>>> at java.lang.Thread.run(Thread.java:701)
>>>>>
>>>>> and
>>>>>
>>>>>> java.lang.NoSuchMethodError:
>>>>>
>>>
>> org.apache.flink.streaming.api.datastream.DataStream.distribute()Lorg/apache/flink/streaming/api/datastream/DataStream;
>>>>>> at
>>>>>
>>>
>> org.apache.flink.stormcompatibility.api.FlinkTopologyBuilder.createTopology(FlinkTopologyBuilder.java:155)
>>>>>> at
>>>>>
>>>
>> org.apache.flink.stormcompatibility.wordcount.StormWordCountLocal.main(StormWordCountLocal.java:66)
>>>>>> at
>>>>>
>>>
>> org.apache.flink.stormcompatibility.wordcount.StormWordCountLocalITCase.testProgram(StormWordCountLocalITCase.java:45)
>>>>>
>>>>> Looking into the code the method "run(...)" is implemented in
>>>>> AbstractStormSpoutWrapper.java what is the base class of
>>>>> StormFiniteSpoutWrapper.java. So I cannot explain this error...
>>>>>
>>>>> see:
>>>>>>
>>>>>
>>>
>> https://github.com/mjsax/flink/blob/18a3c8b79c5f353bbfd65370811e900edc7abc89/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormFiniteSpoutWrapper.java
>>>>>>
>>>>>
>>>
>> https://github.com/mjsax/flink/blob/18a3c8b79c5f353bbfd65370811e900edc7abc89/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/AbstractStormSpoutWrapper.java
>>>>>
>>>>> Furthermore, the method "distribute()" was used in an old version of
>> the
>>>>> layer, but got replaces by "shuffle()". Thus, I don't understand the
>>>>> second error message either...
>>>>>
>>>>> see:
>>>>>>
>>>>>
>>>
>> https://github.com/mjsax/flink/blob/18a3c8b79c5f353bbfd65370811e900edc7abc89/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilder.java
>>>>>
>>>>> Something must be wrong with travis. It seems not to work on the
>> correct
>>>>> code. However, I have no idea why... Maybe some version mixup.
>>>>>
>>>>> And ideas how I can fix this?
>>>>>
>>>>>
>>>>> -Matthias
>>>>>
>>>>>
>>>>>
>>>>
>>>
>>>
>>
>
signature.asc
Description: OpenPGP digital signature
