MINOR: Fix streams Scala peek recursive call#5566
MINOR: Fix streams Scala peek recursive call#5566guozhangwang merged 4 commits intoapache:trunkfrom tedyu:trunk
Conversation
| testDriver.close() | ||
| } | ||
|
|
||
| "peek a KStream" should "side effect records" in { |
There was a problem hiding this comment.
the test name is a bit confusing to me: peek should not have side effect on records right?
There was a problem hiding this comment.
In this test we are creating a side effect since we modify an external var.
Maybe it should be side effect on records?
Anyway if we change the name here we should probably change on foreach too.
There was a problem hiding this comment.
Yeah the name is a bit confusing: I thought it meant that peek / foreach can modify the records themselves, which is not correct. If it is meant for "peek / foreach can have side effects on external variables", any operators like map / filter can have that side effect as well.
There was a problem hiding this comment.
@tedyu What about: should run foreach actions on records?
guozhangwang
left a comment
guozhangwang
left a comment
There was a problem hiding this comment.
cc @vvcephei to take another look.
| acc shouldBe "value1value2" | ||
|
|
||
| testDriver.close() | ||
| } |
There was a problem hiding this comment.
I would do something like this:
"peek a KStream" should "side effect records" in {
val builder = new StreamsBuilder()
val sourceTopic = "source"
val sinkTopic = "sink"
var acc = ""
builder.stream[String, String](sourceTopic).peek((_, v) => acc += v).to(sinkTopic)
val testDriver = createTestDriver(builder)
testDriver.pipeRecord(sourceTopic, ("1", "value1"))
acc shouldBe "value1"
testDriver.readRecord[String, String](sinkTopic).value shouldBe "value1"
testDriver.pipeRecord(sourceTopic, ("2", "value2"))
acc shouldBe "value1value2"
testDriver.readRecord[String, String](sinkTopic).value shouldBe "value2"
testDriver.close()
}
| @@ -91,6 +91,22 @@ class KStreamTest extends FlatSpec with Matchers with TestDriver { | |||
| testDriver.close() | |||
| } | |||
There was a problem hiding this comment.
Could you do the change on this one too please? I forgot to test in between.
testDriver.pipeRecord(sourceTopic, ("1", "value1"))
acc shouldBe "value1"
testDriver.pipeRecord(sourceTopic, ("2", "value2"))
acc shouldBe "value1value2"
|
Minor comments, LGTM 👍 |
|
Thanks for the review, @joan38 I have addressed your comments |
guozhangwang
left a comment
guozhangwang
left a comment
There was a problem hiding this comment.
Other than the naming, lgtm.
| testDriver.close() | ||
| } | ||
|
|
||
| "peek a KStream" should "side effect records" in { |
There was a problem hiding this comment.
Yeah the name is a bit confusing: I thought it meant that peek / foreach can modify the records themselves, which is not correct. If it is meant for "peek / foreach can have side effects on external variables", any operators like map / filter can have that side effect as well.
| val testDriver = createTestDriver(builder) | ||
|
|
||
| testDriver.pipeRecord(sourceTopic, ("1", "value1")) | ||
| acc shouldBe "value1" |
There was a problem hiding this comment.
Sorry for my OCD but could you add a new line after this so that we see it's different steps of the test?
|
@tedyu Just wanted to make sure you saw my comment above #5566 (comment) |
|
Are you referring to comment about side effect ? I saw it. |
|
Yes. WDYT? I'm really bad a naming/phrasing... |
|
I would wait for @guozhangwang to comment. I think the current form should be fine. But I am open to refinement. |
|
@guozhangwang @joan38 thanks |
|
I think we are all good on this one. |
vvcephei
left a comment
vvcephei
left a comment
There was a problem hiding this comment.
It LGTM as well. Thanks @tedyu
/cc @guozhangwang
|
Ping @guozhangwang |
|
The most satisfying part of our work summarised in one word: |
Sorry for the delay, I was OOO yesterday. |
|
@guozhangwang This needs to be cherry picked to 2.0 as well |
|
Yup! Done. |
This PR fixes the previously recursive call of Streams Scala peek Reviewers: Joan Goyeau <joan@goyeau.com>, Guozhang Wang <guozhang@confluent.io>, John Roesler <john@confluent.io>
|
Thanks! |
|
If you don't mind me asking, what is the release/publish plan for this? |
|
It will be in 2.0.1, which it looks like hasn't been planned yet: https://cwiki.apache.org/confluence/display/KAFKA/Future+release+plan |
This PR fixes the previously recursive call of Streams Scala peek Reviewers: Joan Goyeau <joan@goyeau.com>, Guozhang Wang <guozhang@confluent.io>, John Roesler <john@confluent.io>
6 participants
This PR fixes the previously recursive call of Streams Scala peek
Committer Checklist (excluded from commit message)