lucliu1108 commented on code in PR #20968:
URL: https://github.com/apache/kafka/pull/20968#discussion_r2785908610
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java:
##########
@@ -96,27 +98,38 @@ public void writeToTopology(final InternalTopologyBuilder
topologyBuilder) {
false
);
} else {
- topologyBuilder.addSource(consumedInternal().offsetResetPolicy(),
- sourceName,
- consumedInternal().timestampExtractor(),
- consumedInternal().keyDeserializer(),
- consumedInternal().valueDeserializer(),
- topicName);
-
- processorParameters.addProcessorTo(topologyBuilder, sourceName);
-
- // if the KTableSource should not be materialized, stores will be
null or empty
final KTableSource<K, V> tableSource = (KTableSource<K, V>)
processorParameters.processorSupplier();
- if (tableSource.stores() != null) {
- if (shouldReuseSourceTopicForChangelog) {
- // TODO: rewrite this part to use
Topology.addReadOnlyStateStore() instead
- // should allow to move off using
`InternalTopologyBuilder` in favor of the public `Topology` API
- tableSource.stores().forEach(store -> {
- // connect the source topic as (read-only) changelog
topic for fault-tolerance
- store.withLoggingDisabled();
-
topologyBuilder.connectSourceStoreAndTopic(store.name(), topicName);
- });
- }
+ if (tableSource.stores() != null &&
shouldReuseSourceTopicForChangelog) {
+ // The DSL topology uses a serial pipeline,
+ // whereas the PAPI optimization assumes a parallel structure.
+ // We use a wildcard supplier to support the DSL's forwarding
behavior.
+ final ProcessorSupplier<K, V, ?, ?> stateUpdater =
+
processorParameters.wrappedProcessSupplier(topologyBuilder);
+
+ // TableSourceNode is backed by KTableSource, which maintains
a single state store.
+ // Therefore, we are guaranteed to have exactly one store.
+ assert tableSource.stores().size() == 1;
Review Comment:
Could assert fail silently in production code(not completely sure) Looks
like assert isn't used in other Java code.
Might be better to throw an exception instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
