lightbend/kafka-streams-scala

KStreamS#merge is recursive, causing StackOverflow on initialization

pacman899 opened this issue · 3 comments

pacman899 commented

def merge(stream: KStreamS[K, V]): KStreamS[K, V] = inner.merge(stream)

this is the definition of KStreamS#merge which causes stackoverflow. I think the cause is that the inner KStream instance is getting wrapped implicitly with KStreamS and calling its merge recursively. I believe just changing the name of the function should fix that.

from a quick look, a few more functions there has the same issue.

dnrusakov commented

@debasishg Observing the same StackOverflowError when i do stream1.merge(stream2)

java.lang.StackOverflowError
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)
	at com.lightbend.kafka.scala.streams.KStreamS.merge(KStreamS.scala:168)

It maybe the same problem as in here: #41

debasishg commented

Thanks for pointing this out. I have the fix and will release it soon .. I fixed it this way ..

def merge(stream: KStreamS[K, V]): KStreamS[K, V] = inner.merge(stream.inner)
debasishg commented

@pacman899 -

from a quick look, a few more functions there has the same issue.

Can u please point out the other functions where u noted this ?