You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/11 15:16:53 UTC
[GitHub] [kafka] cadonna opened a new pull request #9863: MINOR: Restore interrupt status when closing
cadonna opened a new pull request #9863:
URL: https://github.com/apache/kafka/pull/9863
We do not always own the thread that executes the close() method, i.e.,
we do not know the interruption policy of the thread. Thus, we should
not swallow the interruption. The least we can do is restoring the
interruption status before the current thread exits this method.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r563978428
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
That's right, I don't think force quit uses the interrupt path. But anyways I don't think it's possible to force quit a Kafka Streams application until we get around to making it a desktop application. I was just trying to draw a parallel -- my concern here is just that there is absolutely no way to recover should a StreamThread hang during shutdown, meaning you will just need to kill the process and do an unclean shutdown. With ALOS it's maybe not so bad, but with EOS an unclean shutdown requires wiping out all state stores and completely restoring them from scratch. So there is absolutely something to lose besides just the annoyance of having to manually kill the Streams process.
I just got a notification from a report of Streams being stuck during shutdown in an EOS application, so I promise I'm not making this up 🙂 . I can point you to it if you'd like -- unfortunately we never even managed to determine the root cause, so there may be more bugs out there which cause this besides just the ones we currently know
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562813288
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
Nope, I just forget about this PR -- thanks for the reminder
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#issuecomment-765658341
Two unrelated fail tests:
```
org.apache.kafka.clients.consumer.internals.FetcherTest.testEarlierOffsetResetArrivesLate()
org.apache.kafka.clients.producer.KafkaProducerTest.testHeadersWithExtendedClasses()
``
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman edited a comment on pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman edited a comment on pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#issuecomment-765658341
Two unrelated fail tests:
```
org.apache.kafka.clients.consumer.internals.FetcherTest.testEarlierOffsetResetArrivesLate()
org.apache.kafka.clients.producer.KafkaProducerTest.testHeadersWithExtendedClasses()
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#issuecomment-759120863
By the way, we now also swallow an InterruptedException in one other place: `StreamThread#waitOnThreadState`. Similar pattern, used for `#removeThread` instead of `#close`. Just wanted to bring it up since it seems like we should apply the same principle to that usage, but you may have missed it if you haven't rebased on trunk in the past 24 hours.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562774792
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
@ableegoldman Are there still open questions after our offline sync?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556176651
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
Maybe we should have a quick sync on this. My understanding is that an interrupt means that the thread wants to regain control somewhere along the callstack. So the only way I can see to interpret it is as "(1) stop blocking on this thread, (2) get the system back into a consistent state, and then (3) reset the flag so the interrupt can be handled (or not) by the caller". Before this PR we were doing (1) and (2), now we're doing (2) and (3), but why not all three?
If we don't break out of the loop then we've effectively ignored (but not swallowed) the interrupt, since we will go on waiting for the thread to reach DEAD. Worse, the thread will now be in a busy loop, as you mentioned in another comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555168775
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
Review comment:
do we want to interrupt before we return? if that is the case why wait until the condition is full-filled?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
I am not 100% sure this is the purpose of the interrupt flag but I think that this will do
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
}
- elapsedMs = time.milliseconds() - begin;
}
return true;
Review comment:
if we have this inside the try as well then we should return and set the interrupted status. That should be better for the user if they want to ignore the exception because they can be sure of the state change and avoid making this call again
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r563978428
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
That's right, I don't think force quit uses the interrupt path. But anyways I don't think it's possible to force quit a Kafka Streams application until we get around to making it a desktop application. I was just trying to draw a parallel -- my concern here is just that there is absolutely no way to recover should a StreamThread hang during shutdown, meaning you will just need to kill the process and do an unclean shutdown. With ALOS it's maybe not so bad, but with EOS an unclean shutdown requires wiping out all state stores and completely restoring them from scratch. So there is absolutely something to lose besides just the annoyance of having to manually kill the Streams process.
I just got a notification from a report of Streams being stuck during shutdown in an EOS application, so I promise I'm not making this up 🙂 . I can point you to it if you'd like -- unfortunately we never even managed to determine the root cause, so there may be more bugs out there which cause this besides just the ones we currently know
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555319837
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
Review comment:
We do not really interrupt, we restore the interruption state if the current thread was interrupted. What would be the alternative to restoring before we return?
If we do not wait until the condition is fulfilled, the while loop with the wait degenerates to busy waiting because usually the interruption status is checked at the beginning of the `wait()` method and we will run into the `InterruptedException` in each iteration which would defeat the purpose of the wait(). When the `InterruptedException` is thrown, the interruption status is reset.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman merged pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman merged pull request #9863:
URL: https://github.com/apache/kafka/pull/9863
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#issuecomment-758016527
Call for review: @ableegoldman @wcarlson5
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555322808
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
}
- elapsedMs = time.milliseconds() - begin;
}
return true;
Review comment:
As far as I see, moving the return into the try block would not change anything.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555595987
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
In the words of "Java Concurrency in Practice":
"Thread interruption is a cooperative mechanism for a thread to signal another thread that it should, at its convenience and if it feels like it, stop what it is doing and do something else."
Interruption consists of the interruption status that can be set by the current thread itself or by another thread with `Thread#interrupt()`. The interruption status can be checked with `Thread#isInterrupted()` and it is checked by some blocking methods, most notably `Object#wait()` and `Thread#sleep()`, which -- if the current thread is interrupted -- clear the interruption status and throw an `InterruptedException`.
Usually, interruption is used for cancellation and "something else" means the application of an interruption policy. Since we do not always own the thread that calls `close()`, we do not know the interruption policy. Also if interruption is not used for cancellation, we cannot always know what "something else" means. So, the minimum that we can do is restoring the interrupted status, if the current thread was interrupted so that further up the call stack of the current thread is able to react on the interruption.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562929843
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
That is a fair concern. Based on the interrupt documentation I would hope that force quit doesn't use the interrupt path. As Bruno said
> In the words of "Java Concurrency in Practice":
"Thread interruption is a cooperative mechanism for a thread to signal another thread that it should, at its convenience and if it feels like it, stop what it is doing and do something else."
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556176651
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
Maybe we should have a quick sync on this. My understanding is that an interrupt means that the thread wants to regain control somewhere along the callstack. So the only way I can see to interpret it is as "(1) stop blocking/waiting on this, (2) get the system back into a consistent state, and then (3) reset the flag so the interrupt can be handled (or not) by the caller". Before this PR we were doing (1) and (2), now we're doing (2) and (3), but why not all three?
If we don't break out of the loop then we've effectively ignored (but not swallowed) the interrupt, since we will go on waiting for the thread to reach DEAD. Worse, the thread will now be in a busy loop, as you mentioned in another comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556039219
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
Don't we want to `break` out of the loop here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562929843
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
That is a fair concern. Based on the interrupt documentation I would hope that force quit doesn't use the interrupt path. As Bruno said
> In the words of "Java Concurrency in Practice":
"Thread interruption is a cooperative mechanism for a thread to signal another thread that it should, at its convenience and if it feels like it, stop what it is doing and do something else."
The "at its convenience" is what makes me less concerned
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555595987
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
In the words of "Java Concurrency in Practice":
"Thread interruption is a cooperative mechanism for a thread to signal another thread that it should, at its convenience and if it feels like it, stop what it is doing and do something else."
Interruption consists of the interruption status that can be set by the current thread itself or by another thread with `Thread#interrupt()`. The interruption status can be checked with `Thread#isInterrupted()` and it is checked by some blocking methods, most notably `Object#wait()` and `Thread#sleep()`, which -- if the current thread is interrupted -- clear the interruption status and throw an `InterruptedException`.
The "something else" means the application of an interruption policy. Since we do not always own the thread that calls `close()`, we do not know the interruption policy. So, the minimum that we can do is restoring the interrupted status if the current thread was interrupted so that further up the call stack of the current thread, the possibly existing interruption policy of the current thread can be applied.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555313715
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
Could you elaborate on why you are not 100% sure?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556095824
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
This makes sense. thanks for explaining. I see why previously it was ignored but its probably best to restore it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562882284
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
Maybe I'm just paranoid from going through some incidents where the thread really did never shut down. I would agree we're probably in a bad & unrecoverable state at that point, but hanging indefinitely is not the right answer. What happens when a program freezes on your laptop? I usually just try to force quit it -- the user might want to do the same. But if we never return control of their thread then it's hard for them to know to shut it down
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562815151
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
I am still a bit concerned about `#removeStreamThread` potentially hanging indefinitely...maybe we should add a timeout to this API? If we want to do so, it would need to be soon since the 2.8 release is coming up. WDYT @cadonna @wcarlson5 ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r562819243
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -612,11 +612,22 @@ public void setStreamsUncaughtExceptionHandler(final java.util.function.Consumer
public void waitOnThreadState(final StreamThread.State targetState) {
synchronized (stateLock) {
- while (state != targetState) {
- try {
- stateLock.wait();
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ boolean interrupted = false;
+ try {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
Review comment:
This particular bit would only hang if the thread is not shutting down. If that is happening maybe we are already in an unrecoverable state?
Overall I wouldn't mind having some sort of timeout but I am not very worried about it. @cadonna maybe you have any insights from writing the KIP?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556176651
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
Maybe we should have a quick sync on this. My understanding is that an interrupt means that the thread wants to regain control somewhere along the callstack. So the only way I can see to interpret it is as "(1) stop blocking/waiting on whatever you're doing, (2) get the system back into a consistent state, and then (3) reset the flag so the interrupt can be handled (or not) by the caller". Before this PR we were doing (1) and (2), now we're doing (2) and (3), but why not all three?
If we don't break out of the loop then we've effectively ignored the interrupt, since we will go on waiting for it to reach NOT_RUNNING. But it's actually worse, since as you mentioned in another comment, it'll now be in a busy loop
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r555320421
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
+ }
+ } else {
+ log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
+ return false;
}
- } else {
- log.debug("Cannot transit to {} within {}ms", targetState, waitMs);
- return false;
+ elapsedMs = time.milliseconds() - begin;
+ }
+ } finally {
+ // Make sure to restore the interruption status before returning.
+ // We do not always own the current thread that executes this method, i.e., we do not know the
+ // interruption policy of the thread. The least we can do is restore the interruption status before
+ // the current thread exits this method.
+ if (interrupted) {
+ Thread.currentThread().interrupt();
Review comment:
I thought it was to break threads out loops, I don't know if the flag is actually checked. So will passing it up be useful if the don't retrun
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a change in pull request #9863: MINOR: Restore interrupt status when closing
Posted by GitBox <gi...@apache.org>.
cadonna commented on a change in pull request #9863:
URL: https://github.com/apache/kafka/pull/9863#discussion_r556076041
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -249,20 +249,31 @@ public boolean isValidTransition(final State newState) {
private boolean waitOnState(final State targetState, final long waitMs) {
final long begin = time.milliseconds();
synchronized (stateLock) {
+ boolean interrupted = false;
long elapsedMs = 0L;
- while (state != targetState) {
- if (waitMs > elapsedMs) {
- final long remainingMs = waitMs - elapsedMs;
- try {
- stateLock.wait(remainingMs);
- } catch (final InterruptedException e) {
- // it is ok: just move on to the next iteration
+ try {
+ while (state != targetState) {
+ if (waitMs > elapsedMs) {
+ final long remainingMs = waitMs - elapsedMs;
+ try {
+ stateLock.wait(remainingMs);
+ } catch (final InterruptedException e) {
+ interrupted = true;
Review comment:
I do not think we should break the loop, because we actually do not know what interruption means for the current thread since we do not own it in general. It might be cancellation -- for which breaking the loop might make sense -- but it could also be something else.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org