chia7712 commented on a change in pull request #9887:
URL: https://github.com/apache/kafka/pull/9887#discussion_r557231712
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -989,16 +986,20 @@ private StreamThread createAndAddStreamThread(final long
cacheSizePerThread, fin
public Optional<String> removeStreamThread() {
if (isRunningOrRebalancing()) {
synchronized (changeThreadCount) {
- for (final StreamThread streamThread : threads) {
- if (streamThread.isAlive() &&
(!streamThread.getName().equals(Thread.currentThread().getName()) ||
threads.size() == 1)) {
- streamThread.shutdown();
- if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
-
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+ synchronized (threads) {
+ final Iterator<StreamThread> iter = threads.iterator();
+ while (iter.hasNext()) {
+ final StreamThread streamThread = iter.next();
+ if (streamThread.isAlive() &&
(!streamThread.getName().equals(Thread.currentThread().getName()) ||
threads.size() == 1)) {
+ streamThread.shutdown();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
+
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+ }
+ iter.remove();
+ final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
+ resizeThreadCache(cacheSizePerThread);
+ return Optional.of(streamThread.getName());
}
- threads.remove(streamThread);
Review comment:
Removing element from collection when looping it is error-prone so I
rewrite it by iterator.
----------------------------------------------------------------
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
