You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by BrainLogic <gi...@git.apache.org> on 2016/12/21 01:03:28 UTC
[GitHub] flink pull request #3035: [ FLINK-4905] Kafka test instability IllegalStateE...
GitHub user BrainLogic opened a pull request:
https://github.com/apache/flink/pull/3035
[ FLINK-4905] Kafka test instability IllegalStateException: Client is not started
Root cause of the issue:
`notifyCheckpointComplete` can occur during the cancellation or `runFetchLoop` fail and call `commitOffset` on closed `curatorClient`, so use `CheckpointLock` to close `curatorClient`.
There is a diagram in the jira that describes behaviour of using `Kafka08Fetcher`.
Notes:
1. I don't like approach where `checkPointLock` is leaked into `SourceContext`, this may lead to deadlock.
2. Work with `ZookeeperOffsetHandler` can continue even after the call Kafka08Fetcher.cancel until the `Handler` will not be null.
3. `ZookeeperOffsetHandler` could have `ReadWriteLock` and use `writeLock` only for close operation, but I have doubt, Flink code base does not contain any `ReentrantLocks`. There is possibility to implement such logic without any locks by using lock-free approach.
4. Also in **jdk8**, we have powerful tool `StampedLock`. In which version of Flink we will be able to use **jdk8** features?
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/BrainLogic/flink FLINK-4905_PR
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3035.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3035
----
commit fd5524eb15ec772f95c4168818264c63e45f5784
Author: cube <ae...@gmail.com>
Date: 2016-12-20T23:41:12Z
[FLINK-4905] Kafka test instability IllegalStateException: Client is not started
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
I think we can take a very simple approach here. Many other parts of the code follow the approach to tolerate exceptions thrown during cancellation, or during asynchronous calls on closed operators.
- The client may be closed during shutting down of the Kafka08Fetcher, before even the surrounding KafkaConsumerBase knows that it is closing
- The KafkaConsumerBase tries to commit offsets, sees the exception, and re-throws it since it assumes it.
- We can make the Kafka08Fetcher catch exceptions when committing offsets, and only re-throwing them if the fetcher is still running. That should do the trick.
BTW: There are plans to make the streaming API sources appear to be single threaded, to avoid that sources have to plan for such situations.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3035
Hi @BrainLogic, we're still seeing this exception in tests sometimes, and it'll be great to have this fixed soon.
Please let us know on how you'd like to proceed with the contribution, otherwise if you don't have time, I can also pick it up :-) Thank you!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
I pushed a fix for that to master in e7cda75b8594417559d6aac6229b5893f5459f0f
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by BrainLogic <gi...@git.apache.org>.
Github user BrainLogic commented on the issue:
https://github.com/apache/flink/pull/3035
Thanks for help and explanation.
But in this approach an unlikely bug will be still occurred:
`zkHandler.prepareAndCommitOffsets(offsets);` throws important exception when running is true
then running becomes `false` and we get a swallowed exception in this rare case.
Maybe at least, should log the exception anyway?
I agree that it is not rational to spend time for time consuming refactoring kafka connector of 8 version.
Current version of kafka is 0.10 and hardly, users will widely use 8 version.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
Dropping Java 7 has not happened and there is not yet consensus in the community, so it probably will not happen in the next weeks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by BrainLogic <gi...@git.apache.org>.
Github user BrainLogic commented on the issue:
https://github.com/apache/flink/pull/3035
Distributed systems and multithreading environments make us think in term of logical clock, like Lamport clock, step by step:
Thread1 - fetcher is running `running = true`
Thread2 performs `zkHandler.prepareAndCommitOffsets(offsets)`
Thread2 `running = true` and `zkHandler.prepareAndCommitOffsets(offsets)` throws an exception
Thread1 stop fetcher and change flag `running = false` in normal way without any exception
Thread2 read the flag `running = false` and `return` although, there is a reason of the commit failure that is different from "Client was closed and running= false"
Thread1 fetcher is stopped successfully
There is no exception or any information in log regarding the exception from `zkHandler.prepareAndCommitOffsets(offsets)`
Again this is a rare case when we can lose a root cause of strange behavior - offset will not be committed although there are no any exceptions. Am I wrong?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
@BrainLogic I'll incorporate your comment into a followup commit...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3035
Thank you for working on this @BrainLogic!
First, regarding the approach proposed here:
The approach should be able to fix the `IllegalStateException` we're encountering.
However, I would also want to try to make `ZookeeperOffsetHandler` thread-safe and insensitive to the ordering `commitOffsets` and `close` are called from the outside. Otherwise, from the context of `Kafka08Fetcher` alone, the reason for the new synchronization isn't that obvious, and might result in harder-to-maintain code in the long run. What do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by BrainLogic <gi...@git.apache.org>.
Github user BrainLogic commented on the issue:
https://github.com/apache/flink/pull/3035
I caught yours idea and together with the argument which I mentioned above - users will not extensively use 8 version of kafka connector, I agree with this proposal.
Let me finish this jira, I can create fix based on yours proposal and try to implement test tonight.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3035: [ FLINK-4905] Kafka test instability IllegalStateE...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/3035
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3035
Regarding some of your notes:
- The exposure of the checkpoint lock through `SourceContext` is meant for sources to atomically update their state (e.x. Kafka offsets) with record emitting, with regards to Flink's checkpointing mechanics for exactly once.
- I don't think there has been any solid discussion on migrating to Java 8 yet.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3035
Thank you for picking this up @StephanEwen! I've taken a look at your approach in the local branch, +1 to the approach.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
Bit of background how the error happens:
- The test throws a `SuccessException`
- While being in the finally clause and shutting down the CluratorClient, the containing `Task` has not seen the exception.
- When the commitOffsets() call fails, this overrides the `SuccessException` as the reason why the streaming program terminated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
I would like to pick this fix up. The exception has still occurred a few times for me in the past, and I prefer the above outlined solution, because it adds less locking on cancellation/shutdown, meaning there are fewer implications on deadlocks or long stalls on cancellation/shutdown.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
Quick question: I am not sure if this scenario can happen like that:
> But in this approach an unlikely bug will be still occurred:
zkHandler.prepareAndCommitOffsets(offsets); throws important exception when running is true
then running becomes false and we get a swallowed exception in this rare case.
Maybe at least, should log the exception anyway?
When the exception is thrown in `zkHandler.prepareAndCommitOffsets(offsets)` and `running = true` then the exception does not get swallowed. The `running` flag can become false only after the exception was thrown. If the flag becomes false due to another reason, then there is either another root cause for the failure, or the operator is shutting down anyways. In both cases, it is probably okay to not report the exception.
Please let me know if you think I have overlooked something here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3035: [ FLINK-4905] Kafka test instability IllegalStateExceptio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3035
I this case, the exception would not be logged, true.
It is a very rare corner case that should not affect correctness, and not really distinguishable from the case where an exception is thrown after the source went into finishing mode.
My feeling is to not log here.
- The advantage is that we don't pollute the log with meaningless exception in the common case (many users would be led onto a false track).
- Given that the corner case you described does not affect any expected behavior (the committing action might as well not have happened at all it it were started a few msecs later) it is okay to "swallow" the exception.
- If it does in fact affect the offset committing in more cases, it will surely also occur at another committing attempt during execution.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---