KAFKA-18026: KIP-1112, clean up StatefulProcessorNode #18195
Conversation
ableegoldman
force-pushed
the
KIP-1112-StatefulProcessorNode-cleanup
branch
from
c46f9a4 to
af0931d
Compare
|
@ableegoldman #17881 adds a "triage" label to PRs from non-committers. Turns out this also affect committers if their membership visibility in the ASF GitHub org is not public. I added instructions for setting your membership visibility to public https://github.com/apache/kafka/blob/trunk/.github/workflows/README.md#pr-triage |
ableegoldman
force-pushed
the
KIP-1112-StatefulProcessorNode-cleanup
branch
from
af0931d to
8c061b7
Compare
This comment was marked as outdated.
This comment was marked as outdated.
ableegoldman
force-pushed
the
KIP-1112-StatefulProcessorNode-cleanup
branch
from
35c459a to
a3238e3
Compare
| aggFunctionName, | ||
| new ProcessorParameters<>(aggregateSupplier, aggFunctionName), | ||
| new String[] {storeFactory.storeName()} |
There was a problem hiding this comment.
This is for line 57 above and independent from the PR: just a thought, could we pass in storeFactory.storeName() in the future?
| @@ -1225,7 +1225,8 @@ public <KOut, VOut> KStream<KOut, VOut> process( | |||
| } | |||
|
|
|||
| final String name = new NamedInternal(named).name(); | |||
| final StatefulProcessorNode<? super K, ? super V> processNode = new StatefulProcessorNode<>( | |||
| final ProcessorToStateConnectorNode<? super K, ? super V> processNode = new | |||
| ProcessorToStateConnectorNode<>( | |||
There was a problem hiding this comment.
nit: is newline necessary?
| ); | ||
| builder.addGraphNode(subscriptionSource, subscriptionReceiveNode); | ||
|
|
||
| final KTableValueGetterSupplier<KO, VO> foreignKeyValueGetter = ((KTableImpl<KO, VO, VO>) foreignKeyTable).valueGetterSupplier(); | ||
| final StatefulProcessorNode<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> subscriptionJoinNode = | ||
| new StatefulProcessorNode<>( | ||
| final ProcessorToStateConnectorNode<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> subscriptionJoinNode = |
There was a problem hiding this comment.
Originally I thought I do not need to use the newly introduced ProcessorToStateConnectorNode for this case but only needed for those process/transform with a store list, but thinking that again we probably do not have another way around at the moment..
There was a problem hiding this comment.
I had the same thought actually. I did take a quick look at it but ultimately decided that whether or not it was possible, it would be too much for this one PR. So I'm going to revisit this in a followup PR if it does indeed make sense to do
| @@ -0,0 +1,74 @@ | |||
| /* | |||
There was a problem hiding this comment.
Should we have a unit test class for this new node?
Final cleanup of StatefulProcessorNode after converting all stateful operators to adding state stores via implementing the
#storesmethod. This PR can't be merged until we merge these two open PRs: