KAFKA-7657: Fixing thread state change to instance state change #6018

guozhangwang · 2018-12-09T23:21:36Z

While looking into KAFKA-7657, I found there are a few loopholes in this logic:

We kept a map of thread-name to thread-state and a global-thread state at the KafkaStreams instance-level, in addition to the instance state itself. stateLock is used when accessing the instance state, however when we are in the thread state change callback, we are accessing both the thread-states as well as the instance state at the same time in the callers of setState without a lock, which is vulnerable to concurrent multi-stream threads. The fix is a) introduce a threadStatesLock in addition to the stateLock, which should always be grabbed to modify the thread-states map before the stateLock for modifying the instance level; and we also defer the checking of the instance-level state inside the setState call.
When transiting to state.RUNNING, we check if all threads are either in RUNNING or DEAD state, this is because some threads maybe dead at the rebalance period but we should still proceed to RUNNING if the rest of threads are still transiting to RUNNING.
Added unit test for 2) above. Also simplified another test as a nit change.

Committer Checklist (excluded from commit message)

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

…ndling-thread-dead-state-change

guozhangwang · 2018-12-09T23:21:44Z

@mjsax @bbejeck @vvcephei

guozhangwang · 2018-12-09T23:22:34Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -269,23 +276,6 @@ private boolean setState(final State newState) {
        return true;
    }

-    private boolean setRunningFromCreated() {


There is no logic change: it is just a nit that I moved this function closer to its caller.

guozhangwang · 2018-12-09T23:22:45Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -397,18 +386,21 @@ public void setGlobalStateRestoreListener(final StateRestoreListener globalState
    final class StreamStateListener implements StreamThread.StateListener {
        private final Map<Long, StreamThread.State> threadState;
        private GlobalStreamThread.State globalThreadState;
+        // this lock should always be held before the state lock
+        private final Object threadStatesLock;


This is for 1).

I don't get why we need this? onChange() is synchronized, so how can a race condition happen?

guozhangwang · 2018-12-09T23:23:21Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-                final StreamThread.State newState = (StreamThread.State) abstractNewState;
-                threadState.put(thread.getId(), newState);
-
-                if (newState == StreamThread.State.PARTITIONS_REVOKED && state != State.REBALANCING) {


This is for 1), i.e. doe not check instance-level state here since any access to this should be lock protected.

guozhangwang · 2018-12-09T23:24:08Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -252,6 +252,13 @@ private boolean setState(final State newState) {
                // when the state is already in NOT_RUNNING, its transition to PENDING_SHUTDOWN or NOT_RUNNING (due to consecutive close calls)
                // will be refused but we do not throw exception here, to allow idempotent close calls
                return false;
+            } else if (state == State.REBALANCING && newState == State.REBALANCING) {


This is for 1) as well: we allow REBALANCE -> REBALANCE and RUNNING -> RUNNING because of the deferred check.

Just looking at this, it seems we need to assign oldState = state after we go the lock?

Also below, when calling stateListener.onChange(state, oldState); -- would we need to call stateListener.onChange(newState, oldState); instead? Otherwise, state could change before we do the callback because the lock is released already.

I cannot follow here. Why would this happen? Similar for RUNNING?

Re first two comments: good call, lgtm.

Re last comment: REBALANCING -> REBALANCING is quite normal, note that before this PR we check this before calling setState(State.REBALANCING); so this is prevented, but part of this fix is to move the logic that needs to access the state into a single place (here). Similarly RUNNING -> RUNNING is possible during starting up phase, where we first set the instance state to RUNNING directly to avoid it transit from CREATED -> REBALANCING, and then when threads are starting, it is possible that maybeSetRunning went through and hence calls setState(RUNNING) again.

guozhangwang · 2018-12-09T23:24:36Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

            for (final StreamThread.State state : threadState.values()) {
-                if (state != StreamThread.State.RUNNING) {
+                if (state != StreamThread.State.RUNNING && state != StreamThread.State.DEAD) {


This one-liner is for 2).

vvcephei

Thanks for this fix!

Can you explain why we need the new threadStatesLock instead of just grabbing the stateLock as a precondition to updating the threadState map?

It seems like an optimization to prevent serializing each thread's state update during, eg a rebalance. I suppose this would be nice if there were a large number of threads per instance. Is that what you had in mind?

vvcephei · 2018-12-10T14:33:33Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+
+                    // special case when global thread is dead
+                    if (newState == GlobalStreamThread.State.DEAD && state != State.ERROR && setState(State.ERROR)) {
+                        log.warn("Global thread has died. The instance will be in error state and should be closed.");


Preexisting, but should this be an error log? (seems to be implied by the text of the log)

I agree. Error log level, seems to be more appropriate.

Ack @vvcephei @mjsax

vvcephei · 2018-12-10T14:35:59Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+            stateLock.notifyAll();
+        }
+
+        // we need to call the user customized state listener outside the state lock to avoid potential deadlocks


Does this mean that the state listener might be invoked out of order?
Eg:

RUNNING -> REBALANCING

CREATED -> RUNNING

Maybe this is already accounted for, and it's why we pass both the old and new state?

Yes it is possible that user's callback maybe triggered out of ordering (in practice may be not very likely), this is not ideal but at that time we did not come up with a better idea how to enforce that we always have instance-level state to go CREATED -> RUNNING -> REBALANCING -> RUNNING at the starting up phase.

All ears if you have a better idea :)

Why could we not fix this? The underlying issue seems to be the state transitions of StreamThread -- it allows to go from CREATED -> RUNNING -- if we change this, and we can only go to CREATED -> PARTITION REVOKED we should be able to tackle this issue? (Maybe a different PR to do this though.)

guozhangwang · 2018-12-10T17:50:14Z

It seems like an optimization to prevent serializing each thread's state update during, eg a rebalance. I suppose this would be nice if there were a large number of threads per instance. Is that what you had in mind?

Yes, the main purpose is to have finer granularity in locks (i.e. one for the threads map and global thread summary, and one for the instance-level state), since there are call paths that actually does not access or modify the threads map at all but directly jump on the state and modify it (e.g. transitFromCreatedToRunning).

bbejeck

Thanks for the fix @guozhangwang, LGTM

mjsax

Some questions. Not sure if I fully understand atm.

mjsax · 2018-12-13T13:21:11Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -424,12 +416,13 @@ private void maybeSetError() {
         * If all threads are up, including the global thread, set to RUNNING
         */
        private void maybeSetRunning() {
-            // one thread is running, check others, including global thread
+            // state can be transferred to RUNNING if all threads are either RUNNING or DEAD


Does it make sense to set the instance to RUNNING if all threads are DEAD ?

Note this is triggered only when a thread transited to RUNNING.

When all threads are DEAD, by the time the last thread transited to DEAD the maybeSetError will proceed and the state will transit to ERROR.

mjsax · 2018-12-13T13:49:01Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+
+                    // special case when global thread is dead
+                    if (newState == GlobalStreamThread.State.DEAD && state != State.ERROR && setState(State.ERROR)) {
+                        log.warn("Global thread has died. The instance will be in error state and should be closed.");


I agree. Error log level, seems to be more appropriate.

mjsax · 2018-12-13T13:55:37Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -252,6 +252,13 @@ private boolean setState(final State newState) {
                // when the state is already in NOT_RUNNING, its transition to PENDING_SHUTDOWN or NOT_RUNNING (due to consecutive close calls)
                // will be refused but we do not throw exception here, to allow idempotent close calls
                return false;
+            } else if (state == State.REBALANCING && newState == State.REBALANCING) {


Just looking at this, it seems we need to assign oldState = state after we go the lock?

mjsax · 2018-12-13T13:56:45Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -252,6 +252,13 @@ private boolean setState(final State newState) {
                // when the state is already in NOT_RUNNING, its transition to PENDING_SHUTDOWN or NOT_RUNNING (due to consecutive close calls)
                // will be refused but we do not throw exception here, to allow idempotent close calls
                return false;
+            } else if (state == State.REBALANCING && newState == State.REBALANCING) {


Also below, when calling stateListener.onChange(state, oldState); -- would we need to call stateListener.onChange(newState, oldState); instead? Otherwise, state could change before we do the callback because the lock is released already.

mjsax · 2018-12-13T13:58:23Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -330,8 +320,7 @@ public void setStateListener(final KafkaStreams.StateListener listener) {
        if (state == State.CREATED) {
            stateListener = listener;
        } else {
-            throw new IllegalStateException("Can only set StateListener in CREATED state. " +
-                    "Current state is: " + state);
+            throw new IllegalStateException("Can only set StateListener in CREATED state. Current state is: " + state);


Seems it might be better to get a lock here, too (ie, for the whole method)? Similar in setUncaughtExceptionHandler() and setGlobalStateRestoreListener()?

mjsax · 2018-12-13T14:03:47Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+            stateLock.notifyAll();
+        }
+
+        // we need to call the user customized state listener outside the state lock to avoid potential deadlocks


Why could we not fix this? The underlying issue seems to be the state transitions of StreamThread -- it allows to go from CREATED -> RUNNING -- if we change this, and we can only go to CREATED -> PARTITION REVOKED we should be able to tackle this issue? (Maybe a different PR to do this though.)

mjsax · 2018-12-13T14:05:41Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -252,6 +252,13 @@ private boolean setState(final State newState) {
                // when the state is already in NOT_RUNNING, its transition to PENDING_SHUTDOWN or NOT_RUNNING (due to consecutive close calls)
                // will be refused but we do not throw exception here, to allow idempotent close calls
                return false;
+            } else if (state == State.REBALANCING && newState == State.REBALANCING) {


I cannot follow here. Why would this happen? Similar for RUNNING?

mjsax · 2018-12-13T14:08:22Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -397,18 +386,21 @@ public void setGlobalStateRestoreListener(final StateRestoreListener globalState
    final class StreamStateListener implements StreamThread.StateListener {
        private final Map<Long, StreamThread.State> threadState;
        private GlobalStreamThread.State globalThreadState;
+        // this lock should always be held before the state lock
+        private final Object threadStatesLock;


I don't get why we need this? onChange() is synchronized, so how can a race condition happen?

…ndling-thread-dead-state-change

guozhangwang · 2018-12-14T21:14:19Z

| Why could we not fix this? The underlying issue seems to be the state transitions of StreamThread -- it allows to go from CREATED -> RUNNING -- if we change this, and we can only go to CREATED -> PARTITION REVOKED we should be able to tackle this issue? (Maybe a different PR to do this though.)

Not sure I fully understand, could you elaborate?

guozhangwang · 2018-12-14T21:16:07Z

| I don't get why we need this? onChange() is synchronized, so how can a race condition happen?

There are possible concurrent access to the threadState and globalThreadState, and hence we need to synchronize on that. We can of course just put everything under synchronization of stateLock, but like I mentioned before:

the main purpose is to have finer granularity in locks (i.e. one for the threads map and global thread summary, and one for the instance-level state), since there are call paths that actually does not access or modify the threads map at all but directly jump on the state and modify it (e.g. transitFromCreatedToRunning).

mjsax · 2018-12-15T10:30:27Z

About state transitions. My comment was a little bit ambiguous. It was not about callback order, but about avoiding: CREATED -> RUNNING -> REBALANCING -> RUNNING at startup.

Currently, we have state transitions for StreamThread

     *                +-------------+
     *          +<--- | Created (0) |
     *          |     +-----+-------+
     *          |           |
     *          |           v
     *          |     +-----+-------+
     *          +<--- | Running (1) | <----+
     *          |     +-----+-------+      |
     *          |           |              |
     *          |           v              |
     *          |     +-----+-------+      |
     *          +<--- | Partitions  |      |
     *          |     | Revoked (2) | <----+
     *          |     +-----+-------+      |
     *          |           |              |
     *          |           v              |
     *          |     +-----+-------+      |
     *          |     | Partitions  |      |
     *          |     | Assigned (3)| ---->+
     *          |     +-----+-------+
     *          |           |
     *          |           v
     *          |     +-----+-------+
     *          +---> | Pending     |
     *                | Shutdown (4)|
     *                +-----+-------+
     *                      |
     *                      v
     *                +-----+-------+
     *                | Dead (5)    |
     *                +-------------+

Those lead to CREATED -> RUNNING -> REBALANCING -> RUNNING.

Why not change it to:

     *                +-------------+
     *          +<--- | Created (0) |
     *          |     +-----+-------+
     *          |           |
     *          |           v
     *          |     +-----+-------+
     *          +<--- | Partitions  | <-------+
     *          |     | Revoked (2) | <----+  |
     *          |     +-----+-------+      |  |
     *          |           |              |  |
     *          |           v              |  |
     *          |     +-----+-------+      |  |
     *          +<--- | Partitions  | ---->+  |
     *          |     | Assigned (3)|      |  |
     *          |     +-----+-------+      |  |
     *          |           |              |  |
     *          |           v              |  |
     *          |     +-----+-------+      |  |
     *          +<--- | Running (1) | <----+  |
     *          |     +-----+-------+         |
     *          |           |                 |
     *          |           +-----------------+
     *          |
     *          |     +-------------+
     *          +---> | Pending     |
     *                | Shutdown (4)|
     *                +-----+-------+
     *                      |
     *                      v
     *                +-----+-------+
     *                | Dead (5)    |
     *                +-------------+

This way, on startup we get CREATED -> REBALANCING -> RUNNING for KafkaStreams states.

guozhangwang · 2018-12-17T00:24:28Z

I see. So what I can do is:

on StreamThread.start, setState(PARTITION_REVOKED).
allow stream thread state PARTITION_REVOKED --> PARTITION_REVOKED itself. This is because we need to call setState(PARTITION_REVOKED) inside onPartitionRevoked callback still.
on KafkaStreams.start, remove setRunningFromCreated; just rely on stream thread to set it to REBALANCING instead.

1. still check stream instance state REBALANCING --> REBALANCING as did in the current PR.

Is that right?

I think it is doable, and CREATED -> REBALANCING -> RUNNING does look better. But since it changes the state transition a bit that would require the KIP. If people prefer this as well I can piggy-back this on a small proposal.

…ndling-thread-dead-state-change

guozhangwang · 2018-12-17T04:57:08Z

This is the state transition diagram change: 359c95b cc @vvcephei @mjsax @bbejeck

mjsax · 2018-12-17T12:55:24Z

I would not set PARTITION_REVOKED on StreamThread.start, because we get this transition automatically in onPartitionRevoked.

Of course, this implies that we are in state CREATED a little longer, but this seems to be fine to me personally.

If we need a KIP, I think we should to a separate PR but not piggy-back to this PR.

guozhangwang · 2018-12-17T21:23:10Z

I would not set PARTITION_REVOKED on StreamThread.start, because we get this transition automatically in onPartitionRevoked.

The reason for immediate state transition is to disallow KafkaStreams.start() being called multiple times; we have a unit test in KafkaStreamsTest for this case specifically. Of course we can also add a waitOnState(REBALANCING) inside the start() function after kicking start the threads -- actually I've also implemented this along with this PR -- but the complexity and timing dependency seems even worse to me, and hence I remained in this approach.

If we need a KIP, I think we should to a separate PR but not piggy-back to this PR.

Thinking about this again, I'd agree with you. In fact I'd rather treat it as a minor bug and update the documentations on operations directly.

mjsax · 2018-12-18T10:54:26Z

What about adding one more state with transitions CREATED -> STARTED -> PARTITIONS_REVOKED on startup? (STARTING instead of STARTED)

guozhangwang · 2018-12-18T16:36:17Z

What about adding one more state with transitions CREATED -> STARTED -> PARTITIONS_REVOKED on startup? (STARTING instead of STARTED)

Are you referring to instance state or thread state?

mjsax · 2018-12-18T21:44:40Z

Thread state. Not sure if we need an additional state at the instance level.

…ndling-thread-dead-state-change

guozhangwang · 2018-12-19T08:11:36Z

I see your point now. I've added the STARTING state to stream thread.

mjsax

Overall LGTM.

Some minor follow up questions.

mjsax · 2018-12-19T09:40:03Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+                // when the state is already in REBALANCING, it should not transit to REBALANCING
+                return false;
+            } else if (state == State.RUNNING && newState == State.RUNNING) {
+                // when the state is already in RUNNING, it should not transit to RUNNING


With the state transitions changes for StreamThread is this still possible?

Good point, I can remove it here.

mjsax · 2018-12-19T09:43:40Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-                // special case when global thread is dead
-                if (newState == GlobalStreamThread.State.DEAD && state != State.ERROR && setState(State.ERROR)) {
-                    log.warn("Global thread has died. The instance will be in error state and should be closed.");
+            synchronized (threadStatesLock) {


Why do we need threadStateLock? Can't we use this?

We can; but since we have a stateLock already I want to distinguish it with another dedicated object.

mjsax · 2018-12-19T09:51:04Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+
+                    // special case when global thread is dead
+                    if (newState == GlobalStreamThread.State.DEAD && state != State.ERROR && setState(State.ERROR)) {
+                        log.error("Global thread has died. The instance will be in error state and should be closed.");


If we want to encapsulate state handling at instance/thread level, maybe rewrite this to:

if (newState == GlobalStreamThread.State.DEAD) { setState(State.ERROR); log.error("Global thread has died. The instance will be in error state and should be closed."); }

GlobalStreamThread would call this method only once anyway, and setState() should handle idempotent setState(State.ERROR) internally?

I can do that, but I need to let State.ERROR transit to State.ERROR (currently it is not allowed, and hence will cause illegal state exception). Will do that.

mjsax · 2018-12-19T09:54:14Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -788,17 +789,13 @@ public synchronized void start() throws IllegalStateException, StreamsException

            final Long cleanupDelay = config.getLong(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG);
            stateDirCleaner.scheduleAtFixedRate(() -> {
+                // we do not use lock here since we only read on the value and act on it


Not sure if this holds. Don't we need to block state transitions while we cleanup is running?

stateDirectory.cleanRemovedTasks(cleanupDelay); itself have synchronization barriers as well, so we do not need to have the whole function in synchronized(stateLock); and in that case, just locking stateLock for reading its value does not bring any additional guarantees.

mjsax · 2018-12-19T09:55:45Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-            // since we do not allow calling start multiple times whether or not it is already shutdown.
-            // TODO: In the future if we lift this restriction this code path could then be triggered and be updated
-            log.error("Already stopped, cannot re-start");
+            throw new IllegalStateException("The client is either already started or already stopped, cannot re-start");


Why do we want to disallow calling start() twice? Could be idempotent no-op, too.

Good question.. this was added at the very beginning when we try to fix a few state transition bugs, and one of them as calling start() twice which may re-create threads etc. Arguably we can still allow calling it twice while making second / future calls no-op.

I'd suggest we leave it as a separate improvement.

mjsax · 2018-12-19T09:59:11Z

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

     *         In this case we will forbid the transition but will not treat as an error.
     *     </li>
     * </ul>
     */
    public enum State implements ThreadStateTransitionValidator {
-        CREATED(1, 4), RUNNING(2, 4), PARTITIONS_REVOKED(3, 4), PARTITIONS_ASSIGNED(1, 2, 4), PENDING_SHUTDOWN(5), DEAD;
+        CREATED(1, 5), STARTING(2, 5), PARTITIONS_REVOKED(3, 5), PARTITIONS_ASSIGNED(2, 4, 5), RUNNING(2, 5), PENDING_SHUTDOWN(6), DEAD;


Should is be PARTITIONS_REVOKED(2, 3, 5) because it want to transit to itself?

Okay, I can remove the check in setState and add it here.

EDIT: I've thought about it twice, and I felt it is better to not allow "transiting to itself" but just avoid it in the setState instead. The main reason is that otherwise, the stateListener.onChange will be called which may trigger user's listener and cause confusion (think: why my listener gets called again on error? it was triggered already).

…ndling-thread-dead-state-change

…ge (#6018)" This reverts commit d669830.

…ge (#6018)" (#6090) This reverts commit d669830.

guozhangwang · 2019-01-04T22:44:58Z

Reverted this one as I found it lacks a final commit in my local branch, sorry.

Re-created #6091.

…he#6018) While looking into KAFKA-7657, I found there are a few loopholes in this logic: 1. We kept a map of thread-name to thread-state and a global-thread state at the KafkaStreams instance-level, in addition to the instance state itself. stateLock is used when accessing the instance state, however when we are in the thread state change callback, we are accessing both the thread-states as well as the instance state at the same time in the callers of setState without a lock, which is vulnerable to concurrent multi-stream threads. The fix is a) introduce a threadStatesLock in addition to the stateLock, which should always be grabbed to modify the thread-states map before the stateLock for modifying the instance level; and we also defer the checking of the instance-level state inside the setState call. 2. When transiting to state.RUNNING, we check if all threads are either in RUNNING or DEAD state, this is because some threads maybe dead at the rebalance period but we should still proceed to RUNNING if the rest of threads are still transiting to RUNNING. Added unit test for 2) above. Also simplified another test as a nit change. Reviewers: John Roesler <vvcephei@users.noreply.github.com>, Matthias J. Sax <mjsax@apache.org>

…ge (apache#6018)" (apache#6090) This reverts commit d669830.

guozhangwang added 3 commits December 6, 2018 16:55

first try

1803694

Merge branch 'trunk' of https://github.com/apache/kafka into K7657-ha…

c49068e

…ndling-thread-dead-state-change

adding a thread state lock

76dd198

guozhangwang commented Dec 9, 2018

View reviewed changes

vvcephei approved these changes Dec 10, 2018

View reviewed changes

bbejeck approved these changes Dec 11, 2018

View reviewed changes

mjsax reviewed Dec 13, 2018

View reviewed changes

Merge branch 'trunk' of https://github.com/apache/kafka into K7657-ha…

1eae489

…ndling-thread-dead-state-change

guozhangwang added 4 commits December 16, 2018 16:45

github comments

441163f

Merge branch 'trunk' of https://github.com/apache/kafka into K7657-ha…

30b1162

…ndling-thread-dead-state-change

modify state transition

2ba9bdb

modify state transition

359c95b

guozhangwang added 2 commits December 18, 2018 23:01

Merge branch 'trunk' of https://github.com/apache/kafka into K7657-ha…

1a6f384

…ndling-thread-dead-state-change

github comments

88fa49a

mjsax reviewed Dec 19, 2018

View reviewed changes

guozhangwang added 2 commits January 4, 2019 11:47

Merge branch 'trunk' of https://github.com/apache/kafka into K7657-ha…

a01aa57

…ndling-thread-dead-state-change

github comments

a32c0e8

guozhangwang merged commit d669830 into apache:trunk Jan 4, 2019

guozhangwang added a commit that referenced this pull request Jan 4, 2019

Revert "KAFKA-7657: Fixing thread state change to instance state chan…

c3d933f

…ge (#6018)" This reverts commit d669830.

guozhangwang mentioned this pull request Jan 4, 2019

Revert "KAFKA-7657: Fixing thread state change to instance state change" #6090

Merged

guozhangwang added a commit that referenced this pull request Jan 4, 2019

Revert "KAFKA-7657: Fixing thread state change to instance state chan…

9d54421

…ge (#6018)" (#6090) This reverts commit d669830.

pengxiaolong pushed a commit to pengxiaolong/kafka that referenced this pull request Jun 14, 2019

Revert "KAFKA-7657: Fixing thread state change to instance state chan…

caf97e3

…ge (apache#6018)" (apache#6090) This reverts commit d669830.

KAFKA-7657: Fixing thread state change to instance state change #6018

KAFKA-7657: Fixing thread state change to instance state change #6018

Conversation

guozhangwang commented Dec 9, 2018 • edited Loading

Committer Checklist (excluded from commit message)

guozhangwang commented Dec 9, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Dec 10, 2018

bbejeck left a comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Dec 14, 2018

guozhangwang commented Dec 14, 2018

mjsax commented Dec 15, 2018

guozhangwang commented Dec 17, 2018

guozhangwang commented Dec 17, 2018

mjsax commented Dec 17, 2018

guozhangwang commented Dec 17, 2018

mjsax commented Dec 18, 2018

guozhangwang commented Dec 18, 2018

mjsax commented Dec 18, 2018

guozhangwang commented Dec 19, 2018

mjsax left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang Jan 4, 2019 • edited Loading

Choose a reason for hiding this comment

guozhangwang commented Jan 4, 2019

guozhangwang commented Dec 9, 2018 •

edited

Loading

mjsax left a comment •

edited

Loading

guozhangwang Jan 4, 2019 •

edited

Loading