[FLINK-24651][iteration] Add bounded all-round iteration #13
Conversation
gaoyunhaii
force-pushed
the
i7_add_bounded_all_round
branch
from
eda820a
to
d17dcb1
Compare
gaoyunhaii
force-pushed
the
i7_add_bounded_all_round
branch
from
d17dcb1
to
49c658f
Compare
There was a problem hiding this comment.
Thanks @gaoyunhaii for opening the pr. I have a little question about building the criteria sub-topology.
In the pr you let the "Head" and "Tail" connect together by introducing a fake operator. Maybe we could the "Head" connect to the "Criteria" operator, which returning by the developer. I think it could avoid introducing a fake operator. Correct me if I miss something.
Thanks
flink-ml-iteration/src/main/java/org/apache/flink/iteration/IterationFactory.java
Outdated
Show resolved
Hide resolved
gaoyunhaii
force-pushed
the
i7_add_bounded_all_round
branch
2 times, most recently
from
4ba3606
to
5a8880b
Compare
...n/src/test/java/org/apache/flink/iteration/itcases/BoundedAllRoundStreamIterationITCase.java
Outdated
Show resolved
Hide resolved
|
|
||
| @Test(timeout = 60000) | ||
| public void testSyncVariableOnlyBoundedIteration() throws Exception { | ||
| try (MiniCluster miniCluster = new MiniCluster(createMiniClusterConfiguration(2, 2))) { |
There was a problem hiding this comment.
Maybe you could use a same minicluster for all the test to reduce the time used in the tests.
There was a problem hiding this comment.
Similarly I extract the process of minicluster
There was a problem hiding this comment.
OK. But I think you could create a jira to track this improvement.
| */ | ||
| public class BoundedAllRoundStreamIterationITCase { | ||
|
|
||
| private static BlockingQueue<OutputRecord<Integer>> result = new LinkedBlockingQueue<>(); |
There was a problem hiding this comment.
I also change this test to SharedObject
flink-ml-iteration/src/main/java/org/apache/flink/iteration/IterationFactory.java
Outdated
Show resolved
Hide resolved
|
|
||
| @Test | ||
| public void testTerminationCriteria() throws Exception { | ||
| try (MiniCluster miniCluster = new MiniCluster(createMiniClusterConfiguration(2, 2))) { |
gaoyunhaii
changed the title
gaoyunhaii
force-pushed
the
i7_add_bounded_all_round
branch
from
5a8880b
to
efd697d
Compare
|
Very thanks @guoweiM for the review! will merge~ |
This PR adds the support of the bounded all-round iteration.