ableegoldman commented on a change in pull request #9695:
URL: https://github.com/apache/kafka/pull/9695#discussion_r537955392
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,22 +924,64 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
- streamThread.shutdown();
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ return Optional.empty();
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Removes one stream thread out of the running stream threads from this
Kafka Streams client.
+ * <p>
+ * The removed stream thread is gracefully shut down. This method does not
specify which stream
+ * thread is shut down.
+ * <p>
+ * Since the number of stream threads decreases, the sizes of the caches
in the remaining stream
+ * threads are adapted so that the sum of the cache sizes over all stream
threads equals the total
+ * cache size specified in configuration {@link
StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG}.
+ *
+ * @return name of the removed stream thread or empty if a stream thread
could not be removed because
+ * no stream threads are alive
+ */
+ public Optional<String> removeStreamThread() {
+ if (isRunningOrRebalancing()) {
Review comment:
Same here, let's log a warning
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,22 +924,64 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
- streamThread.shutdown();
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ return Optional.empty();
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Removes one stream thread out of the running stream threads from this
Kafka Streams client.
+ * <p>
+ * The removed stream thread is gracefully shut down. This method does not
specify which stream
+ * thread is shut down.
+ * <p>
+ * Since the number of stream threads decreases, the sizes of the caches
in the remaining stream
+ * threads are adapted so that the sum of the cache sizes over all stream
threads equals the total
+ * cache size specified in configuration {@link
StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG}.
+ *
+ * @return name of the removed stream thread or empty if a stream thread
could not be removed because
+ * no stream threads are alive
+ */
+ public Optional<String> removeStreamThread() {
+ if (isRunningOrRebalancing()) {
+ for (final StreamThread streamThread : threads) {
Review comment:
We need to protect this with a lock or use a thread-safe data structure
for `threads`, otherwise we can get a ConcurrentModificationException if the
user calls addThread and/or removeThread at the same time (on that note let's
add test coverage for this)
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,28 +924,69 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
- streamThread.shutdown();
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ return Optional.empty();
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Removes one stream thread out of the running stream threads from this
Kafka Streams client.
+ *
+ * The removed stream thread is gracefully shut down. This method does not
specify which stream
+ * thread is shut down.
+ *
+ * Since the number of stream threads decreases, the sizes of the caches
in the remaining stream
+ * threads are adapted so that the sum of the cache sizes over all stream
threads equals the total
+ * cache size specified in configuration {@code cache.max.bytes.buffering}.
+ *
+ * @return name of the removed stream thread or empty if a stream thread
could not be removed because
+ * no stream threads are alive
+ */
+ public Optional<String> removeStreamThread() {
+ if (isRunningOrRebalancing()) {
+ for (final StreamThread streamThread : threads) {
+ if (streamThread.isAlive()) {
+ streamThread.shutdown();
+ while (streamThread.state() != StreamThread.State.DEAD &&
!streamThread.getName().equals(Thread.currentThread().getName())) {
Review comment:
I take it the
`!streamThread.getName().equals(Thread.currentThread().getName())` is in
preparation for adding the `REPLACE_THREAD` enum -- if so, can you just leave a
`//TODO` here for now and add this in the followup PR so we have relevant
changes reviewed together?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,22 +924,64 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
- streamThread.shutdown();
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ return Optional.empty();
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Removes one stream thread out of the running stream threads from this
Kafka Streams client.
+ * <p>
+ * The removed stream thread is gracefully shut down. This method does not
specify which stream
+ * thread is shut down.
+ * <p>
+ * Since the number of stream threads decreases, the sizes of the caches
in the remaining stream
+ * threads are adapted so that the sum of the cache sizes over all stream
threads equals the total
+ * cache size specified in configuration {@link
StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG}.
+ *
+ * @return name of the removed stream thread or empty if a stream thread
could not be removed because
+ * no stream threads are alive
+ */
+ public Optional<String> removeStreamThread() {
+ if (isRunningOrRebalancing()) {
+ for (final StreamThread streamThread : threads) {
+ if (streamThread.isAlive()) {
+ streamThread.shutdown();
+ while (streamThread.state() != StreamThread.State.DEAD &&
!streamThread.getName().equals(Thread.currentThread().getName())) {
+ try {
+ synchronized (streamThread.state()) {
+ streamThread.state().wait(100);
+ }
+ } catch (final InterruptedException e) {
+ e.printStackTrace();
Review comment:
Be careful about swallowing the InterruptedException completely. Getting
this exception means the user wants the thread to stop, not just this specific
method. And we don't know what the caller of `removeThread` looks like, it
might just be an infinite loop that checks on some metric and adjusts the
thread count if necessary. We should make sure to propagate the interrupt once
we finish cleaning up after the thread
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,28 +924,69 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
Review comment:
Nice. One minor suggestion, log a warning if the client isn't running
(or rebalancing) and print the current state
----------------------------------------------------------------
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
