kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [kafka] 01/01: Revert "KAFKA-7657: Fixing thread state change to instance state change (#6018)"
Date Fri, 04 Jan 2019 22:40:40 GMT
This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch revert-6018-K7657-handling-thread-dead-state-change
in repository https://gitbox.apache.org/repos/asf/kafka.git

commit c3d933f856a811a2d4f17ee519b35658109c74c8
Author: Guozhang Wang <wangguoz@gmail.com>
AuthorDate: Fri Jan 4 14:40:32 2019 -0800

    Revert "KAFKA-7657: Fixing thread state change to instance state change (#6018)"
    
    This reverts commit d6698308194625e7921b9c3ace27a918f42f26f1.
---
 .../org/apache/kafka/streams/KafkaStreams.java     | 178 +++++++++++----------
 .../streams/processor/internals/StreamThread.java  |  71 ++++----
 .../org/apache/kafka/streams/KafkaStreamsTest.java |  75 ++-------
 .../StreamTableJoinIntegrationTest.java            |   2 +-
 .../integration/utils/IntegrationTestUtils.java    |  10 +-
 .../processor/internals/StreamThreadTest.java      |  50 +++---
 6 files changed, 170 insertions(+), 216 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 4d0c1c7..bbd0105 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -159,24 +159,24 @@ public class KafkaStreams implements AutoCloseable {
      *         |       +-----+--------+
      *         |             |
      *         |             v
-     *         |       +----+--+------+
-     *         |       | Re-          |
-     *         +<----- | Balancing (1)| -------->+
-     *         |       +-----+-+------+          |
-     *         |             | ^                 |
-     *         |             v |                 |
-     *         |       +--------------+          v
-     *         |       | Running (2)  | -------->+
-     *         |       +------+-------+          |
-     *         |              |                  |
-     *         |              v                  |
-     *         |       +------+-------+     +----+-------+
-     *         +-----> | Pending      |<--- | Error (5)  |---+
-     *                 | Shutdown (3) |     +------------+   |
-     *                 +------+-------+          ^           |
-     *                        |                  |           |
-     *                        v                  +-----------+
-     *                 +------+-------+
+     *         |       +--------------+
+     *         +<----- | Running (2)  | -------->+
+     *         |       +----+--+------+          |
+     *         |            |  ^                 |
+     *         |            v  |                 |
+     *         |       +----+--+------+          |
+     *         |       | Re-          |          v
+     *         |       | Balancing (1)| -------->+
+     *         |       +-----+--------+          |
+     *         |             |                   |
+     *         |             v                   v
+     *         |       +-----+--------+     +----+-------+
+     *         +-----> | Pending      |<--- | Error (5)  |
+     *                 | Shutdown (3) |     +------------+
+     *                 +-----+--------+
+     *                       |
+     *                       v
+     *                 +-----+--------+
      *                 | Not          |
      *                 | Running (4)  |
      *                 +--------------+
@@ -191,7 +191,7 @@ public class KafkaStreams implements AutoCloseable {
      *   the instance will be in the ERROR state. The user will need to close it.
      */
     public enum State {
-        CREATED(1, 3), REBALANCING(2, 3, 5), RUNNING(1, 3, 5), PENDING_SHUTDOWN(4), NOT_RUNNING,
ERROR(3, 5);
+        CREATED(2, 3), REBALANCING(2, 3, 5), RUNNING(1, 3, 5), PENDING_SHUTDOWN(4), NOT_RUNNING,
ERROR(3);
 
         private final Set<Integer> validTransitions = new HashSet<>();
 
@@ -238,11 +238,9 @@ public class KafkaStreams implements AutoCloseable {
      * @param newState New state
      */
     private boolean setState(final State newState) {
-        final State oldState;
+        final State oldState = state;
 
         synchronized (stateLock) {
-            oldState = state;
-
             if (state == State.PENDING_SHUTDOWN && newState != State.NOT_RUNNING)
{
                 // when the state is already in PENDING_SHUTDOWN, all other transitions than
NOT_RUNNING (due to thread dying) will be
                 // refused but we do not throw exception here, to allow appropriate error
handling
@@ -251,13 +249,6 @@ public class KafkaStreams implements AutoCloseable {
                 // 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)
{
-                // 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
-                // this can happen during starting up
-//                return false;
             } else if (!state.isValidTransition(newState)) {
                 throw new IllegalStateException("Stream-client " + clientId + ": Unexpected
state transition from " + oldState + " to " + newState);
             } else {
@@ -269,7 +260,24 @@ public class KafkaStreams implements AutoCloseable {
 
         // we need to call the user customized state listener outside the state lock to avoid
potential deadlocks
         if (stateListener != null) {
-            stateListener.onChange(newState, oldState);
+            stateListener.onChange(state, oldState);
+        }
+
+        return true;
+    }
+
+    private boolean setRunningFromCreated() {
+        synchronized (stateLock) {
+            if (state != State.CREATED) {
+                throw new IllegalStateException("Stream-client " + clientId + ": Unexpected
state transition from " + state + " to " + State.RUNNING);
+            }
+            state = State.RUNNING;
+            stateLock.notifyAll();
+        }
+
+        // we need to call the user customized state listener outside the state lock to avoid
potential deadlocks
+        if (stateListener != null) {
+            stateListener.onChange(State.RUNNING, State.CREATED);
         }
 
         return true;
@@ -316,12 +324,11 @@ public class KafkaStreams implements AutoCloseable {
      * @throws IllegalStateException if this {@code KafkaStreams} instance is not in state
{@link State#CREATED CREATED}.
      */
     public void setStateListener(final KafkaStreams.StateListener listener) {
-        synchronized (stateLock) {
-            if (state == State.CREATED) {
-                stateListener = listener;
-            } else {
-                throw new IllegalStateException("Can only set StateListener in CREATED state.
Current state is: " + state);
-            }
+        if (state == State.CREATED) {
+            stateListener = listener;
+        } else {
+            throw new IllegalStateException("Can only set StateListener in CREATED state.
" +
+                    "Current state is: " + state);
         }
     }
 
@@ -333,19 +340,17 @@ public class KafkaStreams implements AutoCloseable {
      * @throws IllegalStateException if this {@code KafkaStreams} instance is not in state
{@link State#CREATED CREATED}.
      */
     public void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler eh) {
-        synchronized (stateLock) {
-            if (state == State.CREATED) {
-                for (final StreamThread thread : threads) {
-                    thread.setUncaughtExceptionHandler(eh);
-                }
+        if (state == State.CREATED) {
+            for (final StreamThread thread : threads) {
+                thread.setUncaughtExceptionHandler(eh);
+            }
 
-                if (globalStreamThread != null) {
-                    globalStreamThread.setUncaughtExceptionHandler(eh);
-                }
-            } else {
-                throw new IllegalStateException("Can only set UncaughtExceptionHandler in
CREATED state. " +
-                    "Current state is: " + state);
+            if (globalStreamThread != null) {
+                globalStreamThread.setUncaughtExceptionHandler(eh);
             }
+        } else {
+            throw new IllegalStateException("Can only set UncaughtExceptionHandler in CREATED
state. " +
+                    "Current state is: " + state);
         }
     }
 
@@ -357,13 +362,11 @@ public class KafkaStreams implements AutoCloseable {
      * @throws IllegalStateException if this {@code KafkaStreams} instance is not in state
{@link State#CREATED CREATED}.
      */
     public void setGlobalStateRestoreListener(final StateRestoreListener globalStateRestoreListener)
{
-        synchronized (stateLock) {
-            if (state == State.CREATED) {
-                this.globalStateRestoreListener = globalStateRestoreListener;
-            } else {
-                throw new IllegalStateException("Can only set GlobalStateRestoreListener
in CREATED state. " +
+        if (state == State.CREATED) {
+            this.globalStateRestoreListener = globalStateRestoreListener;
+        } else {
+            throw new IllegalStateException("Can only set GlobalStateRestoreListener in CREATED
state. " +
                     "Current state is: " + state);
-            }
         }
     }
 
@@ -393,21 +396,18 @@ public class KafkaStreams implements AutoCloseable {
     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;
 
         StreamStateListener(final Map<Long, StreamThread.State> threadState,
                             final GlobalStreamThread.State globalThreadState) {
             this.threadState = threadState;
             this.globalThreadState = globalThreadState;
-            this.threadStatesLock = new Object();
         }
 
         /**
          * If all threads are dead set to ERROR
          */
         private void maybeSetError() {
-            // check if we have at least one thread running
+            // check if we have enough threads running
             for (final StreamThread.State state : threadState.values()) {
                 if (state != StreamThread.State.DEAD) {
                     return;
@@ -415,7 +415,7 @@ public class KafkaStreams implements AutoCloseable {
             }
 
             if (setState(State.ERROR)) {
-                log.error("All stream threads have died. The instance will be in error state
and should be closed.");
+                log.warn("All stream threads have died. The instance will be in error state
and should be closed.");
             }
         }
 
@@ -423,13 +423,12 @@ public class KafkaStreams implements AutoCloseable {
          * If all threads are up, including the global thread, set to RUNNING
          */
         private void maybeSetRunning() {
-            // state can be transferred to RUNNING if all threads are either RUNNING or DEAD
+            // one thread is running, check others, including global thread
             for (final StreamThread.State state : threadState.values()) {
-                if (state != StreamThread.State.RUNNING && state != StreamThread.State.DEAD)
{
+                if (state != StreamThread.State.RUNNING) {
                     return;
                 }
             }
-
             // the global state thread is relevant only if it is started. There are cases
             // when we don't have a global state thread at all, e.g., when we don't have
global KTables
             if (globalThreadState != null && globalThreadState != GlobalStreamThread.State.RUNNING)
{
@@ -444,29 +443,26 @@ public class KafkaStreams implements AutoCloseable {
         public synchronized void onChange(final Thread thread,
                                           final ThreadStateTransitionValidator abstractNewState,
                                           final ThreadStateTransitionValidator abstractOldState)
{
-            synchronized (threadStatesLock) {
-                // StreamThreads first
-                if (thread instanceof StreamThread) {
-                    final StreamThread.State newState = (StreamThread.State) abstractNewState;
-                    threadState.put(thread.getId(), newState);
-
-                    if (newState == StreamThread.State.PARTITIONS_REVOKED) {
-                        setState(State.REBALANCING);
-                    } else if (newState == StreamThread.State.RUNNING) {
-                        maybeSetRunning();
-                    } else if (newState == StreamThread.State.DEAD) {
-                        maybeSetError();
-                    }
-                } else if (thread instanceof GlobalStreamThread) {
-                    // global stream thread has different invariants
-                    final GlobalStreamThread.State newState = (GlobalStreamThread.State)
abstractNewState;
-                    globalThreadState = newState;
-
-                    // special case when global thread is dead
-                    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.");
-                    }
+            // StreamThreads first
+            if (thread instanceof StreamThread) {
+                final StreamThread.State newState = (StreamThread.State) abstractNewState;
+                threadState.put(thread.getId(), newState);
+
+                if (newState == StreamThread.State.PARTITIONS_REVOKED && state !=
State.REBALANCING) {
+                    setState(State.REBALANCING);
+                } else if (newState == StreamThread.State.RUNNING && state != State.RUNNING)
{
+                    maybeSetRunning();
+                } else if (newState == StreamThread.State.DEAD && state != State.ERROR)
{
+                    maybeSetError();
+                }
+            } else if (thread instanceof GlobalStreamThread) {
+                // global stream thread has different invariants
+                final GlobalStreamThread.State newState = (GlobalStreamThread.State) abstractNewState;
+                globalThreadState = newState;
+
+                // 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.");
                 }
             }
         }
@@ -777,9 +773,11 @@ public class KafkaStreams implements AutoCloseable {
      *                          if {@link StreamsConfig#PROCESSING_GUARANTEE_CONFIG exactly-once}
is enabled for pre 0.11.0.x brokers
      */
     public synchronized void start() throws IllegalStateException, StreamsException {
-        if (setState(State.REBALANCING)) {
-            log.debug("Starting Streams client");
+        log.debug("Starting Streams client");
 
+        // first set state to RUNNING before kicking off the threads,
+        // making sure the state will always transit to RUNNING before REBALANCING
+        if (setRunningFromCreated()) {
             if (globalStreamThread != null) {
                 globalStreamThread.start();
             }
@@ -790,13 +788,17 @@ public class KafkaStreams implements AutoCloseable {
 
             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
                 if (state == State.RUNNING) {
                     stateDirectory.cleanRemovedTasks(cleanupDelay);
                 }
             }, cleanupDelay, cleanupDelay, TimeUnit.MILLISECONDS);
+
+            log.info("Started Streams client");
         } else {
-            throw new IllegalStateException("The client is either already started or already
stopped, cannot re-start");
+            // if transition failed but no exception is thrown; currently it is not possible
+            // 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");
         }
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index a4aab98..6c3e7cd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -83,36 +83,30 @@ public class StreamThread extends Thread {
      *          |           |
      *          |           v
      *          |     +-----+-------+
-     *          +<--- | Starting (1)|
-     *          |     +-----+-------+
-     *          |           |
-     *          |           |
-     *          |           v
-     *          |     +-----+-------+
-     *          +<--- | Partitions  |------+
-     *          |     | Revoked (2) | <----+
+     *          +<--- | Running (1) | <----+
      *          |     +-----+-------+      |
      *          |           |              |
      *          |           v              |
      *          |     +-----+-------+      |
-     *          |     | Partitions  |      |
-     *          +<--- | Assigned (3)| ---->+
+     *          +<--- | Partitions  |      |
+     *          |     | Revoked (2) | <----+
      *          |     +-----+-------+      |
      *          |           |              |
      *          |           v              |
      *          |     +-----+-------+      |
-     *          |     | Running (4) | ---->+
+     *          |     | Partitions  |      |
+     *          |     | Assigned (3)| ---->+
      *          |     +-----+-------+
      *          |           |
      *          |           v
      *          |     +-----+-------+
      *          +---> | Pending     |
-     *                | Shutdown (5)|
+     *                | Shutdown (4)|
      *                +-----+-------+
      *                      |
      *                      v
      *                +-----+-------+
-     *                | Dead (6)    |
+     *                | Dead (5)    |
      *                +-------------+
      * </pre>
      *
@@ -127,13 +121,12 @@ public class StreamThread extends Thread {
      *     <li>
      *         State PARTITIONS_REVOKED may want transit to itself indefinitely, in the corner
case when
      *         the coordinator repeatedly fails in-between revoking partitions and assigning
new partitions.
-     *         Also during streams instance start up PARTITIONS_REVOKED may want to transit
to itself as well.
      *         In this case we will forbid the transition but will not treat as an error.
      *     </li>
      * </ul>
      */
     public enum State implements ThreadStateTransitionValidator {
-        CREATED(1, 5), STARTING(2, 5), PARTITIONS_REVOKED(2, 3, 5), PARTITIONS_ASSIGNED(2,
4, 5), RUNNING(2, 5), PENDING_SHUTDOWN(6), DEAD;
+        CREATED(1, 4), RUNNING(2, 4), PARTITIONS_REVOKED(3, 4), PARTITIONS_ASSIGNED(1, 2,
4), PENDING_SHUTDOWN(5), DEAD;
 
         private final Set<Integer> validTransitions = new HashSet<>();
 
@@ -142,7 +135,7 @@ public class StreamThread extends Thread {
         }
 
         public boolean isRunning() {
-            return equals(RUNNING) || equals(STARTING) || equals(PARTITIONS_REVOKED) || equals(PARTITIONS_ASSIGNED);
+            return equals(RUNNING) || equals(PARTITIONS_REVOKED) || equals(PARTITIONS_ASSIGNED);
         }
 
         @Override
@@ -203,10 +196,10 @@ public class StreamThread extends Thread {
                 // when the state is already in NOT_RUNNING, all its transitions
                 // will be refused but we do not throw exception here
                 return null;
-///            } else if (state == State.PARTITIONS_REVOKED && newState == State.PARTITIONS_REVOKED)
{
+            } else if (state == State.PARTITIONS_REVOKED && newState == State.PARTITIONS_REVOKED)
{
                 // when the state is already in PARTITIONS_REVOKED, its transition to itself
will be
                 // refused but we do not throw exception here
-//                return null;
+                return null;
             } else if (!state.isValidTransition(newState)) {
                 log.error("Unexpected state transition from {} to {}", oldState, newState);
                 throw new StreamsException(logPrefix + "Unexpected state transition from
" + oldState + " to " + newState);
@@ -745,7 +738,7 @@ public class StreamThread extends Thread {
     @Override
     public void run() {
         log.info("Starting");
-        if (setState(State.STARTING) == null) {
+        if (setState(State.RUNNING) == null) {
             log.info("StreamThread already shutdown. Not running");
             return;
         }
@@ -823,7 +816,7 @@ public class StreamThread extends Thread {
             // try to fetch some records with normal poll time
             // in order to wait long enough to get the join response
             records = pollRequests(pollTime);
-        } else if (state == State.RUNNING || state == State.STARTING) {
+        } else if (state == State.RUNNING) {
             // try to fetch some records with normal poll time
             // in order to get long polling
             records = pollRequests(pollTime);
@@ -1256,6 +1249,23 @@ public class StreamThread extends Thread {
         return indent + "\tStreamsThread threadId: " + getName() + "\n" + taskManager.toString(indent);
     }
 
+    // the following are for testing only
+    void setNow(final long now) {
+        this.now = now;
+    }
+
+    TaskManager taskManager() {
+        return taskManager;
+    }
+
+    Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> standbyRecords()
{
+        return standbyRecords;
+    }
+
+    int currentNumIterations() {
+        return numIterations;
+    }
+
     public Map<MetricName, Metric> producerMetrics() {
         final LinkedHashMap<MetricName, Metric> result = new LinkedHashMap<>();
         if (producer != null) {
@@ -1290,25 +1300,4 @@ public class StreamThread extends Thread {
         result.putAll(adminClientMetrics);
         return result;
     }
-
-    // the following are for testing only
-    void setNow(final long now) {
-        this.now = now;
-    }
-
-    TaskManager taskManager() {
-        return taskManager;
-    }
-
-    Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> standbyRecords()
{
-        return standbyRecords;
-    }
-
-    int currentNumIterations() {
-        return numIterations;
-    }
-
-    public StreamThread.StateListener stateListener() {
-        return stateListener;
-    }
 }
diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
index 89f3730..d2a4ace 100644
--- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
@@ -139,77 +139,36 @@ public class KafkaStreamsTest {
     }
 
     @Test
-    public void testStateCloseAfterCreate() {
-        globalStreams.close();
-
-        Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, globalStreams.state());
-    }
-
-    @Test
-    public void testStateOneThreadDeadButRebalanceFinish() throws InterruptedException {
+    public void testStateChanges() throws InterruptedException {
         final StateListenerStub stateListener = new StateListenerStub();
         globalStreams.setStateListener(stateListener);
 
-        Assert.assertEquals(0, stateListener.numChanges);
         Assert.assertEquals(KafkaStreams.State.CREATED, globalStreams.state());
+        Assert.assertEquals(0, stateListener.numChanges);
 
         globalStreams.start();
-
         TestUtils.waitForCondition(
-            () -> stateListener.numChanges == 2,
+            () -> globalStreams.state() == KafkaStreams.State.RUNNING,
             10 * 1000,
             "Streams never started.");
-        Assert.assertEquals(KafkaStreams.State.RUNNING, globalStreams.state());
-
-        for (final StreamThread thread: globalStreams.threads) {
-            thread.stateListener().onChange(
-                thread,
-                StreamThread.State.PARTITIONS_REVOKED,
-                StreamThread.State.RUNNING);
-        }
-
-        Assert.assertEquals(3, stateListener.numChanges);
-        Assert.assertEquals(KafkaStreams.State.REBALANCING, globalStreams.state());
-
-        for (final StreamThread thread: globalStreams.threads) {
-            thread.stateListener().onChange(
-                thread,
-                StreamThread.State.PARTITIONS_ASSIGNED,
-                StreamThread.State.PARTITIONS_REVOKED);
-        }
 
-        Assert.assertEquals(3, stateListener.numChanges);
-        Assert.assertEquals(KafkaStreams.State.REBALANCING, globalStreams.state());
-
-        globalStreams.threads[NUM_THREADS - 1].stateListener().onChange(
-            globalStreams.threads[NUM_THREADS - 1],
-            StreamThread.State.PENDING_SHUTDOWN,
-            StreamThread.State.PARTITIONS_ASSIGNED);
+        globalStreams.close();
 
-        globalStreams.threads[NUM_THREADS - 1].stateListener().onChange(
-            globalStreams.threads[NUM_THREADS - 1],
-            StreamThread.State.DEAD,
-            StreamThread.State.PENDING_SHUTDOWN);
+        Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, globalStreams.state());
+    }
 
-        Assert.assertEquals(3, stateListener.numChanges);
-        Assert.assertEquals(KafkaStreams.State.REBALANCING, globalStreams.state());
+    @Test
+    public void testStateCloseAfterCreate() {
+        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
 
-        for (final StreamThread thread: globalStreams.threads) {
-            if (thread != globalStreams.threads[NUM_THREADS - 1]) {
-                thread.stateListener().onChange(
-                    thread,
-                    StreamThread.State.RUNNING,
-                    StreamThread.State.PARTITIONS_ASSIGNED);
-            }
+        try {
+            final StateListenerStub stateListener = new StateListenerStub();
+            streams.setStateListener(stateListener);
+        } finally {
+            streams.close();
         }
 
-        Assert.assertEquals(4, stateListener.numChanges);
-        Assert.assertEquals(KafkaStreams.State.RUNNING, globalStreams.state());
-
-        globalStreams.close();
-
-        Assert.assertEquals(6, stateListener.numChanges);
-        Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, globalStreams.state());
+        Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, streams.state());
     }
 
     @Test
@@ -540,8 +499,8 @@ public class KafkaStreamsTest {
         assertNotNull(threadMetadata);
         assertEquals(2, threadMetadata.size());
         for (final ThreadMetadata metadata : threadMetadata) {
-            assertTrue("#threadState() was: " + metadata.threadState() + "; expected either
RUNNING, STARTING, PARTITIONS_REVOKED, PARTITIONS_ASSIGNED, or CREATED",
-                asList("RUNNING", "STARTING", "PARTITIONS_REVOKED", "PARTITIONS_ASSIGNED",
"CREATED").contains(metadata.threadState()));
+            assertTrue("#threadState() was: " + metadata.threadState() + "; expected either
RUNNING, PARTITIONS_REVOKED, PARTITIONS_ASSIGNED, or CREATED",
+                asList("RUNNING", "PARTITIONS_REVOKED", "PARTITIONS_ASSIGNED", "CREATED").contains(metadata.threadState()));
             assertEquals(0, metadata.standbyTasks().size());
             assertEquals(0, metadata.activeTasks().size());
         }
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java
index ca78d02..7503dd6 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java
@@ -82,7 +82,7 @@ public class StreamTableJoinIntegrationTest extends AbstractJoinIntegrationTest
         TestUtils.waitForCondition(listener::revokedToPendingShutdownSeen, "Did not seen
thread state transited to PENDING_SHUTDOWN");
 
         streams.close();
-        assertEquals(listener.createdToRevokedSeen(), true);
+        assertEquals(listener.runningToRevokedSeen(), true);
         assertEquals(listener.revokedToPendingShutdownSeen(), true);
     }
 
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
index 1097285..8a2122d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
@@ -75,14 +75,14 @@ public class IntegrationTestUtils {
      * Records state transition for StreamThread
      */
     public static class StateListenerStub implements StreamThread.StateListener {
-        boolean startingToRevokedSeen = false;
+        boolean runningToRevokedSeen = false;
         boolean revokedToPendingShutdownSeen = false;
         @Override
         public void onChange(final Thread thread,
                              final ThreadStateTransitionValidator newState,
                              final ThreadStateTransitionValidator oldState) {
-            if (oldState == StreamThread.State.STARTING && newState == StreamThread.State.PARTITIONS_REVOKED)
{
-                startingToRevokedSeen = true;
+            if (oldState == StreamThread.State.RUNNING && newState == StreamThread.State.PARTITIONS_REVOKED)
{
+                runningToRevokedSeen = true;
             } else if (oldState == StreamThread.State.PARTITIONS_REVOKED && newState
== StreamThread.State.PENDING_SHUTDOWN) {
                 revokedToPendingShutdownSeen = true;
             }
@@ -92,8 +92,8 @@ public class IntegrationTestUtils {
             return revokedToPendingShutdownSeen;
         }
 
-        public boolean createdToRevokedSeen() {
-            return startingToRevokedSeen;
+        public boolean runningToRevokedSeen() {
+            return runningToRevokedSeen;
         }
     }
 
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index dd311fb..5df8fbc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -163,12 +163,12 @@ public class StreamThreadTest {
         assertEquals(thread.state(), StreamThread.State.CREATED);
 
         final ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener;
+        thread.setState(StreamThread.State.RUNNING);
 
         final List<TopicPartition> revokedPartitions;
         final List<TopicPartition> assignedPartitions;
 
         // revoke nothing
-        thread.setState(StreamThread.State.STARTING);
         revokedPartitions = Collections.emptyList();
         rebalanceListener.onPartitionsRevoked(revokedPartitions);
 
@@ -202,7 +202,7 @@ public class StreamThreadTest {
         TestUtils.waitForCondition(new TestCondition() {
             @Override
             public boolean conditionMet() {
-                return thread.state() == StreamThread.State.STARTING;
+                return thread.state() == StreamThread.State.RUNNING;
             }
         }, 10 * 1000, "Thread never started.");
 
@@ -342,7 +342,7 @@ public class StreamThreadTest {
             properties));
         final StreamThread thread = createStreamThread(clientId, config, false);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.setState(StreamThread.State.PARTITIONS_REVOKED);
 
         final Set<TopicPartition> assignedPartitions = Collections.singleton(t1p1);
@@ -505,8 +505,8 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, config, false);
 
-        thread.setState(StreamThread.State.STARTING);
-        thread.rebalanceListener.onPartitionsRevoked(Collections.emptyList());
+        thread.setState(StreamThread.State.RUNNING);
+        thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
         final List<TopicPartition> assignedPartitions = new ArrayList<>();
@@ -542,7 +542,7 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)),
true);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -577,7 +577,7 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)),
true);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -612,6 +612,8 @@ public class StreamThreadTest {
     public void shouldShutdownTaskManagerOnClose() {
         final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
         final TaskManager taskManager = EasyMock.createNiceMock(TaskManager.class);
+        EasyMock.expect(taskManager.activeTasks()).andReturn(Collections.<TaskId, StreamTask>emptyMap());
+        EasyMock.expect(taskManager.standbyTasks()).andReturn(Collections.<TaskId, StandbyTask>emptyMap());
         taskManager.shutdown(true);
         EasyMock.expectLastCall();
         EasyMock.replay(taskManager, consumer);
@@ -635,7 +637,7 @@ public class StreamThreadTest {
             new StreamThread.StateListener() {
                 @Override
                 public void onChange(final Thread t, final ThreadStateTransitionValidator
newState, final ThreadStateTransitionValidator oldState) {
-                    if (oldState == StreamThread.State.CREATED && newState == StreamThread.State.STARTING)
{
+                    if (oldState == StreamThread.State.CREATED && newState == StreamThread.State.RUNNING)
{
                         thread.shutdown();
                     }
                 }
@@ -709,8 +711,8 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, config, false);
 
-        thread.setState(StreamThread.State.STARTING);
-        thread.rebalanceListener.onPartitionsRevoked(Collections.emptyList());
+        thread.setState(StreamThread.State.RUNNING);
+        thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
 
         final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
 
@@ -734,7 +736,7 @@ public class StreamThreadTest {
 
         consumer.updatePartitions(topic1, singletonList(new PartitionInfo(topic1, 1, null,
null, null)));
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -802,7 +804,7 @@ public class StreamThreadTest {
         internalTopologyBuilder.addSource(null, "name", null, null, null, topic1);
         internalTopologyBuilder.addSink("out", "output", null, null, null, "name");
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -838,7 +840,7 @@ public class StreamThreadTest {
         internalTopologyBuilder.addSource(null, "name", null, null, null, topic1);
         internalTopologyBuilder.addSink("out", "output", null, null, null, "name");
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -894,7 +896,8 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, config, false);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
+
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -943,7 +946,8 @@ public class StreamThreadTest {
         restoreConsumer.updateEndOffsets(offsets);
         restoreConsumer.updateBeginningOffsets(offsets);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
+
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
@@ -1006,7 +1010,8 @@ public class StreamThreadTest {
             restoreConsumer.addRecord(new ConsumerRecord<>(changelogName2, 1, i, ("K"
+ i).getBytes(), ("V" + i).getBytes()));
         }
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
+
         thread.rebalanceListener.onPartitionsRevoked(null);
 
         final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
@@ -1069,7 +1074,8 @@ public class StreamThreadTest {
 
         final StreamThread thread = createStreamThread(clientId, config, false);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
+
         thread.rebalanceListener.onPartitionsRevoked(null);
         final List<TopicPartition> assignedPartitions = new ArrayList<>();
 
@@ -1115,9 +1121,6 @@ public class StreamThreadTest {
         ThreadMetadata metadata = thread.threadMetadata();
         assertEquals(StreamThread.State.CREATED.name(), metadata.threadState());
 
-        thread.setState(StreamThread.State.STARTING);
-        thread.setState(StreamThread.State.PARTITIONS_REVOKED);
-        thread.setState(StreamThread.State.PARTITIONS_ASSIGNED);
         thread.setState(StreamThread.State.RUNNING);
         metadata = thread.threadMetadata();
         assertEquals(StreamThread.State.RUNNING.name(), metadata.threadState());
@@ -1152,9 +1155,10 @@ public class StreamThreadTest {
 
         clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
 
+        thread.setState(StreamThread.State.RUNNING);
+
         final List<TopicPartition> assignedPartitions = new ArrayList<>();
 
-        thread.setState(StreamThread.State.STARTING);
         thread.rebalanceListener.onPartitionsRevoked(assignedPartitions);
         assertThreadMetadataHasEmptyTasksWithState(thread.threadMetadata(), StreamThread.State.PARTITIONS_REVOKED);
 
@@ -1283,7 +1287,7 @@ public class StreamThreadTest {
         config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
         final StreamThread thread = createStreamThread(clientId, new StreamsConfig(config),
false);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.setState(StreamThread.State.PARTITIONS_REVOKED);
 
         final Set<TopicPartition> assignedPartitions = Collections.singleton(t1p1);
@@ -1327,7 +1331,7 @@ public class StreamThreadTest {
         config.setProperty(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, LogAndSkipOnInvalidTimestamp.class.getName());
         final StreamThread thread = createStreamThread(clientId, new StreamsConfig(config),
false);
 
-        thread.setState(StreamThread.State.STARTING);
+        thread.setState(StreamThread.State.RUNNING);
         thread.setState(StreamThread.State.PARTITIONS_REVOKED);
 
         final Set<TopicPartition> assignedPartitions = Collections.singleton(t1p1);


Mime
View raw message