[
https://issues.apache.org/jira/browse/KAFKA-7055?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16511436#comment-16511436
]
ASF GitHub Bot commented on KAFKA-7055:
---------------------------------------
nixsticks opened a new pull request #5214: KAFKA-7055: Update
InternalTopologyBuilder to throw TopologyException…
URL: https://github.com/apache/kafka/pull/5214
… if a processor or sink is added with no upstream node attached
InternalTopologyBuilder throws an exception if a sink or a processor is
added without at least one upstream node, as records cannot be forwarded
downstream to an unconnected node. This does not prevent users from attempting
to forward to unconnected nodes, but it does prevent them from attaching
effectively useless downstream nodes, and the error message for forwarding to
an unconnected node has been updated to be slightly more specific.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Kafka Streams Processor API allows you to add sinks and processors without
> parent
> ---------------------------------------------------------------------------------
>
> Key: KAFKA-7055
> URL: https://issues.apache.org/jira/browse/KAFKA-7055
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 1.0.0
> Reporter: Nikki Thean
> Assignee: Nikki Thean
> Priority: Minor
> Labels: easyfix
>
> The Kafka Streams Processor API allows you to define a Topology and connect
> sources, processors, and sinks. From reading through the code, it seems that
> you cannot forward a message to a downstream node unless it is explicitly
> connected to the upstream node (from which you are forwarding the message) as
> a child. Here is an example where you forward using name of downstream node
> rather than child index
> ([https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L117]).
> However, I've been able to connect processors and sinks to the topology
> without including parent names, i.e with empty vararg, using this method:
> [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/Topology.java#L423].
> As any attempt to forward a message to those nodes will throw a
> StreamsException, I suggest throwing an exception if a processor or sink is
> added without at least one upstream node. There is a method in
> `InternalTopologyBuilder` that allows you to connect processors by name after
> you add them to the topology, but it is not part of the external Processor
> API.
> In addition (or alternatively), I suggest making [the error message for when
> users try to forward messages to a node that is not
> connected|https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L119]
> more descriptive, like this one for when a user attempts to access a state
> store that is not connected to the processor:
> [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L75-L81]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
