Fix Streams Scala foreach recursive call

[joan38]

Due to lack of conversion to kstream Predicate, existing foreach method in KStream.scala would result in StackOverflowError.

This PR fixes the bug and adds testing for it.

[joan38]

😉

[guozhangwang]

@vvcephei could you take a look?

[joan38]

This PR is good to go.

[vvcephei]

Technically, this is a public interface change, and would require a KIP.

Note that we don't really need the implicit; it just makes the internal implementation nicer. You could instead do:

def foreach(action: (K, V) => Unit): Unit =
    inner.foreach(new ForeachAction[K, V] {
      override def apply(key: K, value: V): Unit = action(key, value)
    })

right?

[joan38]

Yes we could.

[vvcephei]

Can you explain the mechanism by which the call becomes recursive? I feel like I'm missing it...

[vvcephei]

Can you please add a brief description? The description will become the commit message when we squash and merge.

Also, does it make sense to reference KAFKA-7316 also in this PR's title? You could drop a quick comment on that ticket that forEach is also affected.

[joan38]

Basically it's doing an implicit conversion back to the Scala API threw the ImplicitConversions._ import

[joan38]

Now it's not public anymore

[vvcephei]

Ah! I misread this before. This class was already defaulted to public visibility, and we can't retract it now, without a KIP.

I recommend moving the private[streams] to just the new method.

Sorry for the confusion.

( I wish I had noticed before that the whole object was public unnecessarily )

[vvcephei]

Oh, I get it. I'd forgotten that we have (e.g.) implicit def wrapKStream[K, V](inner: KStreamJ[K, V]): KStream[K, V]

[joan38]

Are we good with this one @vvcephei ?

[guozhangwang]

@joan38 last commit causes some conflicts, you'd need to rebase this PR again.

[joan38]

I will, in an hour or so.

[vvcephei]

@joan38 Also, please see my note on privatizing FunctionConversions. I misread it yesterday.

[joan38]

Sure I can do that.

[joan38]

@guozhangwang All good to merge

[vvcephei]

@guozhangwang This needs to be cherry-picked to 2.0

[guozhangwang]

Thanks for the reminder! Just did it.