KAFKA-7316 Use of filter method in KTable.scala may result in StackOverflowError#5543
KAFKA-7316 Use of filter method in KTable.scala may result in StackOverflowError#5543tedyu wants to merge 3 commits intoapache:trunkfrom tedyu:trunk
Conversation
|
@mjsax @guozhangwang |
|
See also: #5538 |
|
Once your PR gets merged, I will adjust mine keeping changes to FunctionConversions.scala, etc which fix peek. |
| */ | ||
| def foreach(action: (K, V) => Unit): Unit = | ||
| inner.foreach((k: K, v: V) => action(k, v)) | ||
| inner.foreach(((k: K, v: V) => action(k, v)).asForeachAction) |
There was a problem hiding this comment.
This can become action.asForeachAction
| */ | ||
| def peek(action: (K, V) => Unit): KStream[K, V] = | ||
| inner.peek((k: K, v: V) => action(k, v)) | ||
| inner.peek(((k: K, v: V) => action(k, v)).asForeachAction) |
| } | ||
| userRegionsTable.filterNot { (_, _) => | ||
| false | ||
| } |
There was a problem hiding this comment.
Once #5543 gets merged, we can implement a proper test a bit like b61cf73#diff-def535526870ef4e9f8b5dcd83b812d5R32.
|
@joan38 Can you update your PR title to reference the JIRA number as they address the same issue. Thx. |
|
@mjsax done |
|
Should I create another JIRA (for peek) since @joan38 has fixed what KAFKA-7316 describes ? |
|
@tedyu Did you just abandoned this PR or you are going to open a new one for |
|
After your PR gets merged, I will create a new one for peek. Thanks |
|
Created #5566 for fixing peek |
4 participants
Due to lack of conversion to kstream Predicate, existing filter method in KTable.scala would result in StackOverflowError.
This PR fixes the bug and adds calls in StreamToTableJoinScalaIntegrationTestImplicitSerdes.testShouldCountClicksPerRegion to prevent regression.
Committer Checklist (excluded from commit message)