From 4fc17ddd9ba00be12c600220d20066973f80fcb3 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 25 Feb 2021 19:22:20 -0800 Subject: [PATCH 1/5] dont remove thread until DEAD, trim later --- .../apache/kafka/streams/KafkaStreams.java | 36 +++++++++++++------ 1 file changed, 26 insertions(+), 10 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 de048d1994011..7c488d5c81e60 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -92,6 +92,7 @@ import java.util.Set; import java.util.TreeMap; import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -463,9 +464,8 @@ private void replaceStreamThread(final Throwable throwable) { closeToError(); } final StreamThread deadThread = (StreamThread) Thread.currentThread(); - threads.remove(deadThread); - addStreamThread(); deadThread.shutdown(); + addStreamThread(); if (throwable instanceof RuntimeException) { throw (RuntimeException) throwable; } else if (throwable instanceof Error) { @@ -1094,16 +1094,32 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout } private int getNextThreadIndex() { - final HashSet names = new HashSet<>(); - processStreamThread(thread -> names.add(thread.getName())); - final String baseName = clientId + "-StreamThread-"; - for (int i = 1; i <= threads.size(); i++) { - final String name = baseName + i; - if (!names.contains(name)) { - return i; + final HashSet allLiveThreadNames = new HashSet<>(); + AtomicInteger maxThreadId = new AtomicInteger(1); + synchronized (threads) { + processStreamThread(thread -> { + // trim any DEAD threads from the list so we can reuse the thread.id + // this is only safe to do once the thread has fully completed shutdown + if (thread.state() == StreamThread.State.DEAD) { + threads.remove(thread); + } else { + allLiveThreadNames.add(thread.getName()); + int threadId = thread.getName().charAt(thread.getName().length() - 1); + if (threadId > maxThreadId.get()) { + maxThreadId.set(threadId); + } + } + }); + + final String baseName = clientId + "-StreamThread-"; + for (int i = 1; i <= maxThreadId.get(); i++) { + final String name = baseName + i; + if (!allLiveThreadNames.contains(name)) { + return i; + } } + return threads.size() + 1; } - return threads.size() + 1; } private long getCacheSizePerThread(final int numStreamThreads) { From 0ee2ef51b9d8c8be5f09e66494125a00418e2a6c Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 25 Feb 2021 19:27:25 -0800 Subject: [PATCH 2/5] dont remove from threads until shut down --- .../main/java/org/apache/kafka/streams/KafkaStreams.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 7c488d5c81e60..1d3830527e2cb 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -1047,9 +1047,15 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout if (!streamThread.waitOnThreadState(StreamThread.State.DEAD, timeoutMs - begin)) { log.warn("Thread " + streamThread.getName() + " did not shutdown in the allotted time"); timeout = true; + // Don't remove from threads until shutdown is complete. We will trim it from the + // list once it reaches DEAD, and if for some reason it's hanging indefinitely in the + // shutdown then we should just consider this thread.id to be burned + } else { + threads.remove(streamThread); } } - threads.remove(streamThread); + // Don't remove from threads until shutdown is complete since this will let another thread + // reuse its thread.id. We will trim any DEAD threads from the list later final long cacheSizePerThread = getCacheSizePerThread(threads.size()); resizeThreadCache(cacheSizePerThread); if (groupInstanceID.isPresent() && callingThreadIsNotCurrentStreamThread) { From 1094754b10828cb18f412d2fd7e78f9da3a103e5 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 25 Feb 2021 19:43:39 -0800 Subject: [PATCH 3/5] fix checkstyle --- checkstyle/suppressions.xml | 2 +- .../src/main/java/org/apache/kafka/streams/KafkaStreams.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 46fb97b3e197e..5d9dd912d1bf5 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -161,7 +161,7 @@ files="StreamThread.java"/> + files="(KafkaStreams|KStreamImpl|KTableImpl).java"/> 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 1d3830527e2cb..8568548105f22 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -1101,7 +1101,7 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout private int getNextThreadIndex() { final HashSet allLiveThreadNames = new HashSet<>(); - AtomicInteger maxThreadId = new AtomicInteger(1); + final AtomicInteger maxThreadId = new AtomicInteger(1); synchronized (threads) { processStreamThread(thread -> { // trim any DEAD threads from the list so we can reuse the thread.id @@ -1110,7 +1110,7 @@ private int getNextThreadIndex() { threads.remove(thread); } else { allLiveThreadNames.add(thread.getName()); - int threadId = thread.getName().charAt(thread.getName().length() - 1); + final int threadId = thread.getName().charAt(thread.getName().length() - 1); if (threadId > maxThreadId.get()) { maxThreadId.set(threadId); } From 6699084f9c9fd64f831348b17f064b7002cee81d Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Mon, 1 Mar 2021 19:34:49 -0800 Subject: [PATCH 4/5] review feedback --- checkstyle/checkstyle.xml | 1 + checkstyle/suppressions.xml | 2 +- .../apache/kafka/streams/KafkaStreams.java | 31 ++++++++++++++----- 3 files changed, 26 insertions(+), 8 deletions(-) diff --git a/checkstyle/checkstyle.xml b/checkstyle/checkstyle.xml index 91045adc60856..7f912dc428a15 100644 --- a/checkstyle/checkstyle.xml +++ b/checkstyle/checkstyle.xml @@ -120,6 +120,7 @@ + diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 5d9dd912d1bf5..46fb97b3e197e 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -161,7 +161,7 @@ files="StreamThread.java"/> + files="(KStreamImpl|KTableImpl).java"/> 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 8568548105f22..f62427d5e125f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -970,7 +970,7 @@ public Optional addStreamThread() { final StreamThread streamThread; synchronized (changeThreadCount) { threadIdx = getNextThreadIndex(); - cacheSizePerThread = getCacheSizePerThread(threads.size() + 1); + cacheSizePerThread = getCacheSizePerThread(getNumLiveStreamThreads() + 1); resizeThreadCache(cacheSizePerThread); // Creating thread should hold the lock in order to avoid duplicate thread index. // If the duplicate index happen, the metadata of thread may be duplicate too. @@ -984,7 +984,7 @@ public Optional addStreamThread() { } else { streamThread.shutdown(); threads.remove(streamThread); - resizeThreadCache(getCacheSizePerThread(threads.size())); + resizeThreadCache(getCacheSizePerThread(getNumLiveStreamThreads())); } } } @@ -1038,7 +1038,7 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout // make a copy of threads to avoid holding lock for (final StreamThread streamThread : new ArrayList<>(threads)) { final boolean callingThreadIsNotCurrentStreamThread = !streamThread.getName().equals(Thread.currentThread().getName()); - if (streamThread.isAlive() && (callingThreadIsNotCurrentStreamThread || threads.size() == 1)) { + if (streamThread.isAlive() && (callingThreadIsNotCurrentStreamThread || getNumLiveStreamThreads() == 1)) { log.info("Removing StreamThread " + streamThread.getName()); final Optional groupInstanceID = streamThread.getGroupInstanceID(); streamThread.requestLeaveGroupDuringShutdown(); @@ -1054,9 +1054,8 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout threads.remove(streamThread); } } - // Don't remove from threads until shutdown is complete since this will let another thread - // reuse its thread.id. We will trim any DEAD threads from the list later - final long cacheSizePerThread = getCacheSizePerThread(threads.size()); + + final long cacheSizePerThread = getCacheSizePerThread(getNumLiveStreamThreads()); resizeThreadCache(cacheSizePerThread); if (groupInstanceID.isPresent() && callingThreadIsNotCurrentStreamThread) { final MemberToRemove memberToRemove = new MemberToRemove(groupInstanceID.get()); @@ -1099,6 +1098,21 @@ private Optional removeStreamThread(final long timeoutMs) throws Timeout return Optional.empty(); } + // Returns the number of threads that are not in the DEAD state -- use this over threads.size() + private int getNumLiveStreamThreads() { + final AtomicInteger numLiveThreads = new AtomicInteger(0); + synchronized (threads) { + processStreamThread(thread -> { + if (thread.state() == StreamThread.State.DEAD) { + threads.remove(thread); + } else { + numLiveThreads.incrementAndGet(); + } + }); + return numLiveThreads.get(); + } + } + private int getNextThreadIndex() { final HashSet allLiveThreadNames = new HashSet<>(); final AtomicInteger maxThreadId = new AtomicInteger(1); @@ -1110,7 +1124,8 @@ private int getNextThreadIndex() { threads.remove(thread); } else { allLiveThreadNames.add(thread.getName()); - final int threadId = thread.getName().charAt(thread.getName().length() - 1); + // Assume threads are always named with the "-StreamThread-" suffix + final int threadId = Integer.parseInt(thread.getName().substring(thread.getName().lastIndexOf("-") + 1)); if (threadId > maxThreadId.get()) { maxThreadId.set(threadId); } @@ -1124,6 +1139,8 @@ private int getNextThreadIndex() { return i; } } + // It's safe to use threads.size() rather than getNumLiveStreamThreads() to infer the number of threads + // here since we trimmed any DEAD threads earlier in this method while holding the lock return threads.size() + 1; } } From 2409f682a5a44fc2c0d5f8684d149229ab174cce Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Tue, 2 Mar 2021 13:36:10 -0800 Subject: [PATCH 5/5] fix test to use accurate thread name --- .../org/apache/kafka/streams/KafkaStreamsTest.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) 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 a4cd8bf22c2ea..b3dd559d88d63 100644 --- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java @@ -232,8 +232,8 @@ private void prepareStreams() throws Exception { EasyMock.expect(StreamThread.processingMode(anyObject(StreamsConfig.class))).andReturn(StreamThread.ProcessingMode.AT_LEAST_ONCE).anyTimes(); EasyMock.expect(streamThreadOne.getId()).andReturn(0L).anyTimes(); EasyMock.expect(streamThreadTwo.getId()).andReturn(1L).anyTimes(); - prepareStreamThread(streamThreadOne, true); - prepareStreamThread(streamThreadTwo, false); + prepareStreamThread(streamThreadOne, 1, true); + prepareStreamThread(streamThreadTwo, 2, false); // setup global threads final AtomicReference globalThreadState = new AtomicReference<>(GlobalStreamThread.State.CREATED); @@ -293,7 +293,7 @@ private void prepareStreams() throws Exception { ); } - private void prepareStreamThread(final StreamThread thread, final boolean terminable) throws Exception { + private void prepareStreamThread(final StreamThread thread, final int threadId, final boolean terminable) throws Exception { final AtomicReference state = new AtomicReference<>(StreamThread.State.CREATED); EasyMock.expect(thread.state()).andAnswer(state::get).anyTimes(); @@ -321,7 +321,7 @@ private void prepareStreamThread(final StreamThread thread, final boolean termin }).anyTimes(); EasyMock.expect(thread.getGroupInstanceID()).andStubReturn(Optional.empty()); EasyMock.expect(thread.threadMetadata()).andReturn(new ThreadMetadata( - "newThead", + "processId-StreamThread-" + threadId, "DEAD", "", "", @@ -337,7 +337,7 @@ private void prepareStreamThread(final StreamThread thread, final boolean termin EasyMock.expectLastCall().anyTimes(); thread.requestLeaveGroupDuringShutdown(); EasyMock.expectLastCall().anyTimes(); - EasyMock.expect(thread.getName()).andStubReturn("newThread"); + EasyMock.expect(thread.getName()).andStubReturn("processId-StreamThread-" + threadId); thread.shutdown(); EasyMock.expectLastCall().andAnswer(() -> { supplier.consumer.close(); @@ -564,7 +564,7 @@ public void shouldAddThreadWhenRunning() throws InterruptedException { streams.start(); final int oldSize = streams.threads.size(); TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L, "wait until running"); - assertThat(streams.addStreamThread(), equalTo(Optional.of("newThread"))); + assertThat(streams.addStreamThread(), equalTo(Optional.of("processId-StreamThread-" + 2))); assertThat(streams.threads.size(), equalTo(oldSize + 1)); } @@ -613,7 +613,7 @@ public void shouldRemoveThread() throws InterruptedException { final int oldSize = streams.threads.size(); TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L, "Kafka Streams client did not reach state RUNNING"); - assertThat(streams.removeStreamThread(), equalTo(Optional.of("newThread"))); + assertThat(streams.removeStreamThread(), equalTo(Optional.of("processId-StreamThread-" + 1))); assertThat(streams.threads.size(), equalTo(oldSize - 1)); }