From 8374c13c94cd50ca1c3cbb3c123b0b841d5ff7bc Mon Sep 17 00:00:00 2001 From: Armin Braun Date: Sun, 2 Apr 2017 12:29:26 +0200 Subject: [PATCH 1/2] MINOR: Fix Deadlock in StreamThread --- .../kafka/clients/producer/KafkaProducer.java | 2 +- .../apache/kafka/streams/KafkaStreams.java | 37 +++++++++++-------- .../processor/internals/StreamThread.java | 8 ++-- .../kafka/streams/KafkaStreamsTest.java | 2 +- 4 files changed, 27 insertions(+), 22 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index 78dd66847e12e..03d91760812e0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/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 firstException = new AtomicReference(); + AtomicReference firstException = new AtomicReference<>(); boolean invokedFromCallback = Thread.currentThread() == this.ioThread; if (timeout > 0) { if (invokedFromCallback) { 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 2c116d92382ec..3de672f44506f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/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, 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 46704b9014a62..3266c90d34f62 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 @@ -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, 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 eebbde95943d0..efa484e9db253 100644 --- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java +++ b/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(); From 05ed2d5946459f1c9cec4f6b9252fa288a952990 Mon Sep 17 00:00:00 2001 From: Armin Date: Sun, 2 Apr 2017 23:16:30 +0100 Subject: [PATCH 2/2] MINOR: Codestyle Fixes --- .../src/main/java/org/apache/kafka/streams/KafkaStreams.java | 5 ++--- .../kafka/streams/processor/internals/StreamThread.java | 2 +- 2 files changed, 3 insertions(+), 4 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 3de672f44506f..6ddf2a1840015 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -220,11 +220,10 @@ public void setStateListener(final KafkaStreams.StateListener listener) { } private void setState(final State newState) { - synchronized (this.stateLock) { + synchronized (stateLock) { final State oldState = state; if (!state.isValidTransition(newState)) { - log.warn( - "{} Unexpected state transition from {} to {}.", logPrefix, oldState, newState); + log.warn("{} Unexpected state transition from {} to {}.", logPrefix, oldState, newState); } else { log.info("{} State transition from {} to {}.", logPrefix, oldState, newState); } 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 3266c90d34f62..9791a0a3bea8c 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 @@ -463,7 +463,7 @@ private RuntimeException performOnAllTasks(final AbstractTaskAction action, action.apply(task); } catch (RuntimeException t) { log.error("{} Failed while executing {} {} due to {}: ", - this.logPrefix, + logPrefix, task.getClass().getSimpleName(), task.id(), exceptionMessage,