KAFKA-6647 KafkaStreams.cleanUp creates .lock file in directory it tries to clean #4702

tedyu · 2018-03-13T18:27:36Z

Specify StandardOpenOption#DELETE_ON_CLOSE when creating the FileChannel.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…ies to clean

gbloggs · 2018-03-13T20:36:26Z

Whilst this may be a nice to have I remain unconvinced that this will fix the issue.
Looking at clean() in the StateDirectory, it catches any IOExceptions thrown from Utils.delete(). Utils.delete in postVisitDirectory will throw java.nio.file.DirectoryNotEmptyException so a new StreamsException(e) will be thrown

public synchronized void clean() {
        try {
            cleanRemovedTasks(0, true);
        } catch (final Exception e) {
            // this is already logged within cleanRemovedTasks
            throw new StreamsException(e);
        }
        try {
            Utils.delete(globalStateDir().getAbsoluteFile());
        } catch (final IOException e) {
            log.error("{} Failed to delete global state directory due to an unexpected exception", logPrefix(), e);
            throw new StreamsException(e);
        }
    }

mjsax · 2018-03-14T05:40:17Z

\cc @bbejeck @vvcephei

bbejeck · 2018-03-14T18:29:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

@@ -138,7 +138,7 @@ synchronized boolean lock(final TaskId taskId) throws IOException {
        }

        try {
-            lockFile = new File(directoryForTask(taskId), LOCK_FILE_NAME);
+            lockFile = new File(stateDir, taskId + LOCK_FILE_NAME);


why remove the directoryForTask method?

The location of the lock file is lifted one level up.

After discussion on #4713 I think this idea should actually work. Nit: Can we rename the lock to LOCK_FILE_NAME + " -" + taskId though.

I can rename the lock if people think we can continue with this solution.

bbejeck · 2018-03-14T18:32:54Z

Left one minor comment. Plus do we need to add a new test?

tedyu · 2018-03-14T19:37:29Z

This seems related:
https://stackoverflow.com/questions/2479222/problem-with-java-file-locking-mechanism-filelock-etc

guozhangwang

I left a comment on https://issues.apache.org/jira/browse/KAFKA-6647 ticket itself.

mjsax · 2018-03-19T22:01:06Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java

@@ -88,27 +89,6 @@ public void shouldCreateTaskStateDirectory() {
        assertTrue(taskDirectory.isDirectory());
    }

-    @Test
-    public void shouldLockTaskStateDirectory() throws IOException {


Why did you remove this test?

directory.lock(taskId) is at different level than channel.tryLock() , rendering the test useless.

tedyu · 2018-03-19T23:01:28Z

Guozhang has different opinion on the approach.
Waiting for discussion to conclude before taking further action.

gartho · 2018-06-18T13:47:12Z

Any update on this issue?

guozhangwang · 2018-06-18T22:09:26Z

@gartho We are still pondering what is the best fix for this issue across different file systems. What's your use case and is this a blocker issue for your app?

gartho · 2018-06-19T05:38:08Z

I’m coming across this issue during integration level testing. I’ve decided to resolve it for now by clearing out the state store folder before I run my tests so it’s not blocking me at present.

philippkamps · 2018-07-11T09:31:57Z

Any updates?

guozhangwang · 2018-07-25T04:57:45Z

@gartho @philippkamps Sorry for being late on this PR. @tedyu I'd suggest follow the option 2) I've proposed in the JIRA ticket, i.e.

Just another note, for locking purposes, when the apps shutdown cleanly, it is OK to have the lock file left in the state directory since threads are excluding each other not via the ownership of the file but via locking the file handle. So for your case, if you indeed want to clean up the whole directory upon shutting down, then I think there is still a valid point to close all the file channels upon shutting down. For which we can consider:

1) either use StandardOpenOption.DELETE_ON_CLOSE as you did in the PR.
2) or add a new function in the state directory class that closes all the managed file channels upon KafkaStreams.close(); which may be safer than 1) since StandardOpenOption.DELETE_ON_CLOSE is best-effort.

WDYT?

pgrabarsky · 2018-08-09T08:29:36Z

+1

mjsax · 2018-09-07T02:01:05Z

@tedyu What is the status of this PR?

tedyu · 2018-09-07T02:15:26Z

From @guozhangwang 's comment on the JIRA, I thought he acknowledged that lifting the lock up one dir would solve the problem for Windows environment.

mjsax · 2018-09-07T02:23:26Z

I just lost the overview :) Last comment from @guozhangwang (#4702 (comment)) seems to indicate that he wants some updates to the PR. Or is this already contained and the PR is ready for reviewing?

Maybe you can rebase the PR to resolve the merge conflict first? Thanks a lot!

tedyu · 2018-09-07T03:08:51Z

Can we discuss how to achieve compatibility with existing locking structure first ?

Thanks

mjsax · 2018-09-07T05:09:40Z

Sure. However, I am not exactly sure what you mean? Can you elaborate?

tedyu · 2018-09-07T16:35:02Z

I meant that, after the move of the lock file, is there any compatibility issue we need to consider.

tedyu · 2018-09-09T16:01:53Z

I didn't find the local repo for the original PR.

Need to open a new one if the approach in current PR is confirmed.

mjsax · 2018-09-13T17:57:44Z

@tedyu Sorry for late reply.

I meant that, after the move of the lock file, is there any compatibility issue we need to consider.

I don't think so. If an app is shut down, it should free all locks. On startup, it can just use the new locking strategy. Or do I miss anything? It would only be an issue, if multiple thread (or different versions) would operate in parallel -- but this case should be be possible on a single machine and thus we should be fine?

tedyu · 2018-09-13T23:57:53Z

bq. but this case should be be possible on a single machine

I guess you meant 'should not be'

tedyu · 2018-09-14T00:10:23Z

Created #5650

guozhangwang · 2018-09-14T18:08:17Z

@tedyu @mjsax My concern about compatibility is that if we have multiple instances of Streams running on the same machine, say A and B, where A shuts down and upgrades to a new version, it will then try to create the lock file in a different directory (one layer up) while B is still holding locks on the old directory, is it possible that both instances think they now own the task's directory and move on to modify the state (e.g. one still owning the task while the other moves in and "cleanup" the state)?

mjsax · 2018-09-14T18:12:18Z

If you run two KafkaStreams instanced on one machine, you need to configure it with two different state.dir -- thus, both should be isolated and the case you describe cannot happen. Or am I wrong with the assumption about different state.dirs ? IIRC, we consider using the same state.dir a configuration error.

bbejeck · 2018-09-14T21:04:48Z

I just started up a small example app that merely created the state.dir. I ran two instances of the application.

In one case I used the same app-id, and both instances started up fine. It makes sense as the different instances will end up writing to different task-id directories.

In the other case, I changed the app-id of the second instance and a new directory state.dir/app-id/<task-directories>.

So I think we need more testing to cover the scenario @guozhangwang outlined above. But it wouldn't hurt to follow @mjsax suggestion to configure with different state.dir entries, but it would be easy to forget to do so.

guozhangwang · 2018-09-15T00:37:04Z

Today we do not actually enforce different state.dir configs for different instances: imagine if Streams is specified via K8s, where the config values are wired in and are the same. K8s may allocate containers on the same machine.

Thoughts @vvcephei @bbejeck @tedyu @dguy ?

tedyu · 2018-09-15T15:43:57Z

I think for K8s deployment, the state directory should be customized for each streaming app.

bhavesh3184 · 2018-09-27T16:53:27Z

@tedyu What is the status of this PR?

tedyu · 2018-09-27T16:57:23Z

See the above comment from Guozhang.
It seems we should consider the case where state.dir is shared between streaming apps

mjsax · 2018-10-03T23:49:42Z

Can we move the discussion to the new PR? Thanks.

KAFKA-6647 KafkaStreams.cleanUp creates .lock file in directory it tr…

4933768

…ies to clean

tedyu added 2 commits March 13, 2018 15:51

Put lock file above task dir

5c8f746

Address checkstyle error

587517e

mjsax added the streams label Mar 14, 2018

mjsax requested review from dguy, mjsax and guozhangwang March 14, 2018 05:40

bbejeck reviewed Mar 14, 2018

View reviewed changes

Modify StateDirectoryTest

e7a5e47

Modify StateDirectoryTest

9e96cf4

guozhangwang reviewed Mar 14, 2018

View reviewed changes

mjsax mentioned this pull request Mar 14, 2018

MINOR: fix KafkaStreams#cleanUp(): should throw and fail for lock conflict #4713

Closed

mjsax reviewed Mar 19, 2018

View reviewed changes

tedyu closed this Sep 14, 2018

sachabarber mentioned this pull request Dec 16, 2018

Another question for you bbejeck/kafka-streams-in-action#2

Open

KAFKA-6647 KafkaStreams.cleanUp creates .lock file in directory it tries to clean #4702

KAFKA-6647 KafkaStreams.cleanUp creates .lock file in directory it tries to clean #4702

Conversation

tedyu commented Mar 13, 2018

Committer Checklist (excluded from commit message)

gbloggs commented Mar 13, 2018

mjsax commented Mar 14, 2018

bbejeck Mar 14, 2018

Choose a reason for hiding this comment

tedyu Mar 14, 2018

Choose a reason for hiding this comment

mjsax Mar 19, 2018

Choose a reason for hiding this comment

tedyu Aug 8, 2018

Choose a reason for hiding this comment

bbejeck commented Mar 14, 2018

tedyu commented Mar 14, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

mjsax Mar 19, 2018

Choose a reason for hiding this comment

tedyu Aug 8, 2018

Choose a reason for hiding this comment

tedyu commented Mar 19, 2018

gartho commented Jun 18, 2018

guozhangwang commented Jun 18, 2018

gartho commented Jun 19, 2018

philippkamps commented Jul 11, 2018

guozhangwang commented Jul 25, 2018

pgrabarsky commented Aug 9, 2018

mjsax commented Sep 7, 2018

tedyu commented Sep 7, 2018

mjsax commented Sep 7, 2018

tedyu commented Sep 7, 2018

mjsax commented Sep 7, 2018

tedyu commented Sep 7, 2018

tedyu commented Sep 9, 2018

mjsax commented Sep 13, 2018

tedyu commented Sep 13, 2018

tedyu commented Sep 14, 2018

guozhangwang commented Sep 14, 2018

mjsax commented Sep 14, 2018 • edited Loading

bbejeck commented Sep 14, 2018

guozhangwang commented Sep 15, 2018

tedyu commented Sep 15, 2018

bhavesh3184 commented Sep 27, 2018

tedyu commented Sep 27, 2018

mjsax commented Oct 3, 2018

mjsax commented Sep 14, 2018 •

edited

Loading