Repository: kafka Updated Branches: refs/heads/trunk 57da044a9 -> 4adfd7960
MINOR: Reorder StreamThread shutdown sequence We need to close producer first before closing tasks to make sure all messages are acked and hence checkpoint offsets are updated before closing tasks and their state. It was re-ordered mistakenly before. Author: Guozhang Wang <[email protected]> Reviewers: Yasuhiro Matsuda <[email protected]>, Ewen Cheslack-Postava <[email protected]> Closes #845 from guozhangwang/KStreamState Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/4adfd796 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/4adfd796 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/4adfd796 Branch: refs/heads/trunk Commit: 4adfd7960c4bef187454d1ff5186f3be690abbf5 Parents: 57da044 Author: Guozhang Wang <[email protected]> Authored: Mon Feb 1 17:32:36 2016 -0800 Committer: Ewen Cheslack-Postava <[email protected]> Committed: Mon Feb 1 17:32:36 2016 -0800 ---------------------------------------------------------------------- .../internals/ProcessorStateManager.java | 3 --- .../processor/internals/StreamThread.java | 22 ++++++++++++-------- 2 files changed, 13 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/4adfd796/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java index bc7f4b5..b90af48 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java @@ -335,9 +335,6 @@ public class ProcessorStateManager { checkpoint.write(checkpointOffsets); } - // un-assign the change log partition - restoreConsumer.assign(Collections.<TopicPartition>emptyList()); - // release the state directory directoryLock directoryLock.release(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/4adfd796/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java ---------------------------------------------------------------------- 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 eccd02c..8948fc8 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 @@ -261,20 +261,16 @@ public class StreamThread extends Thread { private void shutdown() { log.info("Shutting down stream thread [" + this.getName() + "]"); - // We need to first remove the tasks before shutting down the underlying clients - // as they may be required in the previous steps; and exceptions should not - // prevent this call from going through all shutdown steps. try { commitAll(); } catch (Throwable e) { // already logged in commitAll() } - try { - removeStreamTasks(); - removeStandbyTasks(); - } catch (Throwable e) { - // already logged in removeStreamTasks() and removeStandbyTasks() - } + + // We need to first close the underlying clients before closing the state + // manager, for example we need to make sure producer's message sends + // have all been acked before the state manager records + // changelog sent offsets try { producer.close(); } catch (Throwable e) { @@ -291,6 +287,14 @@ public class StreamThread extends Thread { log.error("Failed to close restore consumer in thread [" + this.getName() + "]: ", e); } + // Exceptions should not prevent this call from going through all shutdown steps + try { + removeStreamTasks(); + removeStandbyTasks(); + } catch (Throwable e) { + // already logged in removeStreamTasks() and removeStandbyTasks() + } + log.info("Stream thread shutdown complete [" + this.getName() + "]"); }
