Skip to content

KAFKA-12375: don't reuse thread.id until a thread has fully shut down #10215

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ableegoldman merged 5 commits into from
Mar 3, 2021
Merged

KAFKA-12375: don't reuse thread.id until a thread has fully shut down #10215

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4fc17dd
dont remove thread until DEAD, trim later
ableegoldman Feb 26, 2021
0ee2ef5
dont remove from threads until shut down
ableegoldman Feb 26, 2021
1094754
fix checkstyle
ableegoldman Feb 26, 2021
6699084
review feedback
ableegoldman Mar 2, 2021
2409f68
fix test to use accurate thread name
ableegoldman Mar 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions checkstyle/checkstyle.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@
<module name="ClassDataAbstractionCoupling">
<!-- default is 7 -->
<property name="max" value="25"/>
<property name="excludeClassesRegexps" value="AtomicInteger"/>
</module>
<module name="BooleanExpressionComplexity">
<!-- default is 3 -->
Expand Down
69 changes: 54 additions & 15 deletions streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -463,9 +464,8 @@ private void replaceStreamThread(final Throwable throwable) {
closeToError();
}
final StreamThread deadThread = (StreamThread) Thread.currentThread();
threads.remove(deadThread);

@wcarlson5 wcarlson5 Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I remember that we had the replace use the same ID for a reason. (maybe it had to do with rebalancing?). I don't think there should be a problem to try to get the same ID by waiting a bit in the replace thread

@cadonna cadonna Feb 26, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We cannot wait here until the dead thread is shutdown because the shutdown happens after replaceStreamThread() throws the exception. So we would wait forever.

@wcarlson5 wcarlson5 Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

deadThread.shutdown(); I was referring to this below. But if we don't need to keep the same for any reason I am fine either way

@cadonna cadonna Mar 1, 2021 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for not being clear enough. I was also referring to deadThread.shutdown(). Method deadThread.shutdown() only requests a shutdown. The actual shutdown is performed in completeShutdown() which is called after replaceStreamThread() throws one of the exceptions below. Since completeShutdown() is called by the same thread that calls deadThread.shutdown() in this method, i.e., the dead thread, we would wait forever if we waited after deadThread.shutdown() until the dead thread is shut down. Or did I misunderstand your statement "by waiting a bit in the replace thread"?

@wcarlson5 wcarlson5 Mar 1, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah that makes sense. Another approach is using a temporary name. And then waiting the new thread until the old thread dies and takes the name. This is a bit complicated and I think it should only be done if it is necessary for the new thread to have the same name. And probably not in this PR but it could be an improvement done later

@ableegoldman ableegoldman Mar 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I think swapping the names would make the code unnecessarily complicated, and it would definitely make reading the logs more difficult.
Just to note: in the current rebalance protocol, the thread name should not impact the task assignment since within a client tasks are always just assigned to their previous owner (we maximize stickiness & balance)

@wcarlson5 wcarlson5 Mar 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

since the name doesn't matter I wonder why we spent so much effort making sure it had the same name?

addStreamThread();
deadThread.shutdown();
addStreamThread();
if (throwable instanceof RuntimeException) {
throw (RuntimeException) throwable;
} else if (throwable instanceof Error) {
Expand Down Expand Up @@ -970,7 +970,7 @@ public Optional<String> 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.
Expand All @@ -984,7 +984,7 @@ public Optional<String> addStreamThread() {
} else {
streamThread.shutdown();
threads.remove(streamThread);
resizeThreadCache(getCacheSizePerThread(threads.size()));
resizeThreadCache(getCacheSizePerThread(getNumLiveStreamThreads()));
}
}
}
Expand Down Expand Up @@ -1038,7 +1038,7 @@ private Optional<String> 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<String> groupInstanceID = streamThread.getGroupInstanceID();
streamThread.requestLeaveGroupDuringShutdown();
Expand All @@ -1047,10 +1047,15 @@ private Optional<String> 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

@wcarlson5 wcarlson5 Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Where do we trim this list? I don't thing we do. In the begging of addStreamThread() can we purge the dead threads? That is the only place it should matter

@ableegoldman ableegoldman Mar 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We're trimming it in getNextThreadIndex. But if we're going to rely on threads.size() elsewhere, which it seems we do, then yeah we should trim it more aggressively

// shutdown then we should just consider this thread.id to be burned
} else {
threads.remove(streamThread);

@wcarlson5 wcarlson5 Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we purge the dead threads before we add new ones and if we remove the assumption that there are no dead threads in the thread list we can just not remove the threads in remove thread. This will make it there should be no concern about the cache size changing when a thread is removing itself. And make the risk we took about memory overflows unnecessary.

@ableegoldman ableegoldman Mar 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I was mainly trying to keep things simple. There's definitely a tradeoff in when we resize the cache: either we resize it right away and risk an OOM or we resize it whenever we find newly DEAD threads but potentially have to wait to "reclaim" the memory of a thread.
Both scenarios run into trouble when a thread is hanging in shutdown, but if that occurs something has already gone wrong so I don't think we need to guarantee Streams will continue running perfectly. But the downside to resizing the cache only once a thread reaches DEAD is that a user could call removeStreamThread() with a timeout of 0 and then never call add/remove thread again, and they'll never get back the memory of the removed thread since we only trim the threads inside these methods (or the exception handler). ie, it seems ok to lazily remove DEAD threads if we only use the threads list to find a unique threadId, but not to lazily resize the cache. WDYT?

@wcarlson5 wcarlson5 Mar 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can leave it for now, if we should see problems this could be a fix, we don't run a single thread soak so we won't see this issue ourselves but there are many single thread applications that could start using this and we should see if they have problems

}
}
threads.remove(streamThread);
final long cacheSizePerThread = getCacheSizePerThread(threads.size());

final long cacheSizePerThread = getCacheSizePerThread(getNumLiveStreamThreads());
resizeThreadCache(cacheSizePerThread);
if (groupInstanceID.isPresent() && callingThreadIsNotCurrentStreamThread) {
final MemberToRemove memberToRemove = new MemberToRemove(groupInstanceID.get());
Expand Down Expand Up @@ -1093,17 +1098,51 @@ private Optional<String> 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<String> 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<String> allLiveThreadNames = new HashSet<>();
final AtomicInteger maxThreadId = new AtomicInteger(1);

@cadonna cadonna Feb 26, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we really need an atomic integer here? maxThreadId is only used in the synchronized block.

@ableegoldman ableegoldman Mar 1, 2021 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's because of the whole "variables used in a lambda must be final or effectively final" thing

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());
// Assume threads are always named with the "-StreamThread-<threadId>" suffix
final int threadId = Integer.parseInt(thread.getName().substring(thread.getName().lastIndexOf("-") + 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;
}
}
// 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;
}
return threads.size() + 1;
}

private long getCacheSizePerThread(final int numStreamThreads) {
Expand Down
14 changes: 7 additions & 7 deletions streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<GlobalStreamThread.State> globalThreadState = new AtomicReference<>(GlobalStreamThread.State.CREATED);
Expand Down Expand Up @@ -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<StreamThread.State> state = new AtomicReference<>(StreamThread.State.CREATED);
EasyMock.expect(thread.state()).andAnswer(state::get).anyTimes();

Expand Down Expand Up @@ -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",
"",
"",
Expand All @@ -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();
Expand Down Expand Up @@ -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));
}

Expand Down Expand Up @@ -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));
}

Expand Down