Hi Trey, I think there is a ticket open requesting to be able to re-use the source topic, so I don't think it's an intentional restriction, just a consequence of the way the code is structured at the moment.
Is it sufficient to send the update to "calls" and "answered-calls" at the same time? You could do something like: val answeredCalls = actions.filter { _, action -> action == Actions.ANSWER } .join(callsTable) { id, call -> call } // now a KTable .mapValues { call -> doAnswer(call) } // actual answer implementation answeredCalls.to("calls"); answeredCalls.to("answered-calls"); Does that help? -John On Fri, Feb 15, 2019 at 4:18 PM Trey Hutcheson <trey.hutche...@gmail.com> wrote: > For context, imagine I'm building an IVR simulator. Desired workflow: > > IVR knows about a ringing call. IVR receives an IPC instruction to answer > the call. That instruction is realized by sending a message {action=ANSWER} > to the "actions" topic. > > At this point, the system needs to do two things: actually answer the call, > and then start a recording of the call, in that order. Because of > implementation peculiarities external to the system, assume that these two > things cannot be executed together atomically. > > So this is what I'd *like* to do (warning, kotlin code, types omitted for > brevity): > > val callsTable = builder.table("calls", ...) > val actions = builder.stream("actions", ..) > > actions.filter { _, action -> action == Actions.ANSWER } > .join(callsTable) { id, call -> call } // now a KTable > .mapValues { call -> doAnswer(call) } // actual answer implementation > .through("calls") // persist in state store > .to("answered-calls") // let other actors in the system know the call was > answered, such as start the recording process > > Now in the current version of the streams library (2.1.0), that little bit > of topology throws an exception when trying to build it, with a message > that a source has already been defined for the "calls" topic. So apparently > the call to .through materializes a view and defines a source, which was > already defined in the call to builder.table("calls")? > > So how do I do what I want? This sequence needs to happen in order. I have > tried .branch, but that just ends up in a race condition (the thing doing > to recording has to join to calls table and filter that the call has been > answered). > > I could create a custom processor that forwards to both sinks - but does > that really solve the problem? And if it did, how do I create a > KafkaStreams instance from a combination of StreamBuilder and Topology? > > Thanks for the insight > Trey >