ableegoldman commented on a change in pull request #10355:
URL: https://github.com/apache/kafka/pull/10355#discussion_r597215093
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1097,13 +1102,17 @@ private int getNumStreamThreads(final boolean
hasGlobalTopology) {
return Optional.empty();
}
- // Returns the number of threads that are not in the DEAD state -- use
this over threads.size()
+ // Returns the number of threads that are not in the DEAD or
PENDING_SHUTDOWN 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) {
+ log.debug("Trimming thread {} from the threads list since
it's state is {}", thread.getName(), StreamThread.State.DEAD);
threads.remove(thread);
+ } else if (thread.state() ==
StreamThread.State.PENDING_SHUTDOWN) {
Review comment:
Yep -- but imo this should be pretty rare. To hit an OOM you would need
to have a relatively huge cache, and few total threads (so that the cache size
per thread is still large), and then you'd need the new thread to start up,
rebalance to get tasks, finish initializing and restoring them and then start
consuming enough data to fill up the cache all before the old thread gets
around to closing it's state stores.
Sure, threads can sometimes hang during termination but in my experience
that's usually after the store closure (and hopefully hanging is itself rare).
My point is, this should be good enough for now, and I'd like to stick with
the simplest solution this late in the game.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
