MINOR: Fix Deadlock in StreamThread

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -701,7 +701,7 @@ private void close(long timeout, TimeUnit timeUnit, boolean swallowException) {
 
         log.info("Closing the Kafka producer with timeoutMillis = {} ms.", timeUnit.toMillis(timeout));
         // this will keep track of the first encountered exception
-        AtomicReference<Throwable> firstException = new AtomicReference<Throwable>();
+        AtomicReference<Throwable> firstException = new AtomicReference<>();
         boolean invokedFromCallback = Thread.currentThread() == this.ioThread;
         if (timeout > 0) {
             if (invokedFromCallback) {

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

@@ -189,8 +189,12 @@ public boolean isValidTransition(final State newState) {
             return validTransitions.contains(newState.ordinal());
         }
     }
+
+    private final Object stateLock = new Object();
+
     private volatile State state = State.CREATED;
-    private StateListener stateListener = null;
+
+    private KafkaStreams.StateListener stateListener = null;
 
 
     /**
@@ -208,25 +212,26 @@ public interface StateListener {
     }
 
     /**
-     * An app can set a single {@link StateListener} so that the app is notified when state changes.
+     * An app can set a single {@link KafkaStreams.StateListener} so that the app is notified when state changes.
      * @param listener a new state listener
      */
-    public void setStateListener(final StateListener listener) {
+    public void setStateListener(final KafkaStreams.StateListener listener) {
         stateListener = listener;
     }
 
-    private synchronized void setState(final State newState) {
-        final State oldState = state;
-        if (!state.isValidTransition(newState)) {
-            log.warn("{} Unexpected state transition from {} to {}.", logPrefix, oldState, newState);
-        } else {
-            log.info("{} State transition from {} to {}.", logPrefix, oldState, newState);
-        }
-
-        state = newState;
-
-        if (stateListener != null) {
-            stateListener.onChange(state, oldState);
+    private void setState(final State newState) {
+        synchronized (this.stateLock) {
+            final State oldState = state;
+            if (!state.isValidTransition(newState)) {
+                log.warn(
+                    "{} Unexpected state transition from {} to {}.", logPrefix, oldState, newState);
+            } else {
+                log.info("{} State transition from {} to {}.", logPrefix, oldState, newState);
+            }
+            state = newState;
+            if (stateListener != null) {
+                stateListener.onChange(state, oldState);
+            }
         }
     }
 
@@ -248,7 +253,7 @@ public synchronized State state() {
         return Collections.unmodifiableMap(metrics.metrics());
     }
 
-    private class StreamStateListener implements StreamThread.StateListener {
+    private final class StreamStateListener implements StreamThread.StateListener {
         @Override
         public synchronized void onChange(final StreamThread thread,
                                           final StreamThread.State newState,

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

@@ -124,7 +124,7 @@ public boolean isValidTransition(final State newState) {
     }
 
     private volatile State state = State.NOT_RUNNING;
-    private StateListener stateListener = null;
+    private StreamThread.StateListener stateListener = null;
 
     /**
      * Listen to state change events
@@ -141,10 +141,10 @@ public interface StateListener {
     }
 
     /**
-     * Set the {@link StateListener} to be notified when state changes. Note this API is internal to
+     * Set the {@link StreamThread.StateListener} to be notified when state changes. Note this API is internal to
      * Kafka Streams and is not intended to be used by an external application.
      */
-    public void setStateListener(final StateListener listener) {
+    public void setStateListener(final StreamThread.StateListener listener) {
         this.stateListener = listener;
     }
 
@@ -463,7 +463,7 @@ private RuntimeException performOnAllTasks(final AbstractTaskAction action,
                 action.apply(task);
             } catch (RuntimeException t) {
                 log.error("{} Failed while executing {} {} due to {}: ",
-                        StreamThread.this.logPrefix,
+                        this.logPrefix,
                         task.getClass().getSimpleName(),
                         task.id(),
                         exceptionMessage,

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -282,7 +282,7 @@ public void testCannotCleanupWhileRunning() throws Exception {
         try {
             streams.cleanUp();
         } catch (final IllegalStateException e) {
-            Assert.assertEquals("Cannot clean up while running.", e.getMessage());
+            assertEquals("Cannot clean up while running.", e.getMessage());
             throw e;
         } finally {
             streams.close();