[FLINK-35355][State] Internal async aggregating state and corresponding state descriptor #24810
Conversation
| import java.io.Serializable; | ||
|
|
||
| /** Test implementation of Aggregate function */ | ||
| public class SumAggregator implements AggregateFunction<Integer, Integer, Integer>, Serializable { |
There was a problem hiding this comment.
I'd suggest this to be an inner class if only for testing purpose.
There was a problem hiding this comment.
But it was an inner class, but it is used in both AggregatingStateDescriptorTest and InternalAggregatingStateTest, so I created this class to reuse the code.
There was a problem hiding this comment.
Option 1: This could be a static package-private class within AggregatingStateDescriptorTest reused in both tests. This is common when some test reference the inner static class/variable from another test.
Option 2: Two test inner classes. This makes sense if it is private.
Also, anonymous class is better than inner class.
The reason why I'd suggest make it inner/anonymous is that, typically functional interfaces like this vary in different scenarios, so there's nothing to share. I'd prefer option 2 with anonymous class, since making these tests self contained looks better to me.
There was a problem hiding this comment.
I see, I made it inner class in latest commit. For AggregatingStateDescriptorTest, I found it doesn't need a real instance, so I mocked it.
There was a problem hiding this comment.
It seems flink doesn't encourage the mock, will change it later.
| public class InternalAggregatingState<K, IN, ACC, OUT> extends InternalKeyedState<K, ACC> | ||
| implements AggregatingState<IN, OUT> { | ||
|
|
||
| private final AggregateFunction<IN, ACC, OUT> aggregateFunction; |
There was a problem hiding this comment.
Better be protected since the sub-classes will leverage this function.
There was a problem hiding this comment.
Makes sense to me, will fix it in next commit.
|
Hi @Zakelly, Thanks for the comments, I have added a commit for addressing it. |
jectpro7
force-pushed
the
f35355
branch
3 times, most recently
from
544cfb5
to
a1a0d30
Compare
|
Hi @Zakelly, I've addressed comments, please help double review it. Thanks in advance. |
|
@jectpro7 Seems the author of this commit is different from the committer? Do you mind change that? Or I can merge this as it is. |
|
@Zakelly, It looks weird, seems the git configuration is not correct. I will fix that. |
jectpro7
force-pushed
the
f35355
branch
2 times, most recently
from
9f8b359
to
45fab29
Compare
|
@Zakelly, looks good now. Thanks for the reminder. 👍 |
|
@flinkbot run azure |
1 similar comment
|
@flinkbot run azure |
…ng state descriptor
|
CI green, merging.... |
This PR defines the V2
AggreatingStateand itsStateDescriptor. And provides the defaultInternalAggregatingStateimplementation which delegates all aycn request to AEC.What is the purpose of the change
Supporting aggregate functionality regarding to FLIP-424.
Brief change log
AggregatingStateandAggregatingStateDescriptorAggregatingStateVerifying this change
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation