KAFKA-7367: Streams should not create state store directories unless they are needed#5696
KAFKA-7367: Streams should not create state store directories unless they are needed#5696mjsax merged 7 commits intoapache:trunkfrom
Conversation
|
retest this please |
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Thanks for the PR. And sorry for late review.
I think we can simplify the PR, if we put all the logic into StateDirectory itself.
There was a problem hiding this comment.
Do we need this? Might be simpler to update StateDirectory to not acquire the lock() but just return true for this case?
There was a problem hiding this comment.
Assume that there are multiple tasks in a topology in which some of them have persistent state stores and others have in-memory stores. In this case, StateDirectory don't have any knowledge whether to create the TaskId directory or not.
There was a problem hiding this comment.
Maybe I am missing something, but if there is no stateful task, we don't create the StateDirectory -- and task directories are nested within StateDirectory:
StateDirectory == /<state.dir/<application.id>/
overall:
/<state.dir>/<application.id>/<task.id>/<store>
From my understanding, the ticket addresses the case that there is no disk. If there is disk, and there is one stateful task, we can also create task directories from stateless task -- I agree that it would be ok to not create them, but this increase code complexity and the gain seems to be small?
Thoughts?
There was a problem hiding this comment.
@mjsax I was thinking the same thing. I understand the goal of not creating the directories for stateless tasks, but I'm not sure the increase in code buys much.
There was a problem hiding this comment.
Updated the code to create task directory for all the tasks even if any one of the task has persistent store.
I've done a manual test by running |
|
@bbejeck Would it be sufficient, to run a regular integration test with only stateless tasks and check for the non-existence of the state directory? What do we gain if we test task migration scenario? Seems I am missing something? |
|
Test cases are passing in my local machine. |
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Thanks for updating the PR. Couple of follow up comments.
|
@mjsax |
Yes I'll agree with you, maybe I was being overly conservative. But my point was if a stateful task migrates to to an instance with only stateless tasks previously, we can ensure the directories are created. |
There was a problem hiding this comment.
super nit: maybe change the name as we don't use the EmbeddedBroker anymore? I don't have any good suggestions ATM though.
There was a problem hiding this comment.
Renamed to StatelessTopologyDiskAccessTest and moved to package org.apache.kafka.streams.
There was a problem hiding this comment.
Renamed to TopologyDiskAccessTest to convey the test information clearly.
|
The failures are related, both Java 8 and Java 10 have |
That is a very good point @bbejeck -- I think we should be fine though. In Does this sound correct? |
Sorry about not sharing my thoughts in text. Yes after going over the PR again, I agree with that the |
|
failure unrelated retest this please |
|
retest this please |
There was a problem hiding this comment.
I think we should prefix this with TestUtils.IO_TMP_DIR ?
There was a problem hiding this comment.
We should add a test into ProcessorTopologyTest for this new method
There was a problem hiding this comment.
As above: we should add a test for this method
There was a problem hiding this comment.
We should add a new test to KafkaStreamsTest to check that no directory is created for this case
There was a problem hiding this comment.
For the above case, TopologyDiskAccessTest class is added. KafkaStreamsTest tests only the KafkaStreams logic (It doesn't create any topology which is required for the above case).
There was a problem hiding this comment.
KafkaStreamsTest tests only the KafkaStreams logic
Well, because KafkaStreams is responsible to create/or-not-crate the directory, this is part of KafkaStreams logic, isn't it? Thus, IMHO we should test this behavior in KafkaStreamsTest. Nothing prevents you create any topology in the test for this.
Also, TopologyDisAccessTest does not use KafkaStreams but TopologyTestDriver and thus, does not cover this code path.
There was a problem hiding this comment.
Agree, moved the test cases to KafkaStreamsTest.
There was a problem hiding this comment.
To ensure that state directory should not gets created when using the stateless / in-memory topology with the PAPI / DSL APIs.
There was a problem hiding this comment.
Shouldn't this be part of KafkaStreamsTest (cf. my other comment)
There was a problem hiding this comment.
Removed TopologyDiskAccessTest class.
|
@mjsax |
|
@mjsax |
|
retest this please |
There was a problem hiding this comment.
We should split this into multiple smaller tests, each testing one scenario:
statelessTopologyShouldNotHavePersistentLocalStore()inMemoryStoreShouldNotResultInPersistentLocalStore()persistenLocalStoreShouldBeDetected()
Or similar test method names that explain what the test is doing / expecting / testing.
There was a problem hiding this comment.
Updated the test case.
There was a problem hiding this comment.
Updated the test case.
There was a problem hiding this comment.
Isn't this case covered by statelessPAPITopologyShouldNotCreateStateDirectory ? DSL compiles down into a Topology anyway.
Similar below.
There was a problem hiding this comment.
DSL tests are removed.
There was a problem hiding this comment.
nit: statelessPAPITopologyShouldNotCreateStateDirectory -> statelessTopologyShouldNotCreateStateDirectory
There was a problem hiding this comment.
renamed the test cases.
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Call for second review @guozhangwang @bbejeck @vvcephei
|
Couple for Streams test failures. Do we have tickets for those flaky tests? |
|
@vvcephei @bbejeck @guozhangwang |
|
@kamalcph Could you rebase this PR to resolve the merge conflicts? |
|
@mjsax After rebase, couple of test cases are failing: Class org.apache.kafka.streams.KafkaStreamsTest inMemoryStatefulTopologyShouldNotCreateStateDirectory This one, I'll fix. This may be due to the recent changes in trunk. statelessTopologyShouldNotCreateStateDirectory This case passes when running the test in standalone mode. But, running in test-suite it always fails with InterruptedException. I've tried to debug the issue by pre-creating the directory but the case still fails. Can you help me out to resolve this case ? |
|
The build is green, thus I am a little confused about your statement. I also checked out your PR and run |
bbejeck
left a comment
bbejeck
left a comment
There was a problem hiding this comment.
Overall looks good to me, once the testing is squared away I'll review again.
I'll also try running tests locally and see what I get
|
I also checked out the PR and ran |
|
I'm running the command Today ran the same command thrice. The following test case: |
|
That's interesting. Do you develop on Mac/Linux/Windows? |
|
MAC. $ gradle -version Gradle 4.10Build time: 2018-08-27 18:35:06 UTC Kotlin DSL: 1.0-rc-3 $ java -version |
|
My environment is almost the same. But cannot reproduce... I assume it's only an issue on your machine. Not sure why. Thus, I guess we can safely merge this? Thoughts @bbejeck? |
Ok when I tried this command
they both pass. I think this is ok, as when I have multiple modules to test, I run the tests one at a time. The Jenkins build does the same I believe so I think it's ok to merge. |
|
@mjsax ^^ forgot to add you to the previous comment. |
|
Merged to |
…they are needed (apache#5696) * KAFKA-7367: Ensure stateless topologies don't require disk access * KAFKA-7367: Streams should not create state store directories unless they are needed. * Addressed the review comments. * Addressed the review-2 comments. * Fixed FileAlreadyExistsException * Addressed the review-3 comments. * Resolved the conflicts.
4 participants
More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.
Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.
Committer Checklist (excluded from commit message)