wcarlson5 commented on a change in pull request #9695:
URL: https://github.com/apache/kafka/pull/9695#discussion_r540351596
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,25 +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()));
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
+ }
+ }
+ }
+ log.warn("Cannot add a stream thread in state " + state());
+ 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();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
Review comment:
Maybe there is a global thread and a stream thread and the user only
wants a global thread now?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,25 +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()));
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
+ }
+ }
+ }
+ log.warn("Cannot add a stream thread in state " + state());
+ 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();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
+
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+ }
+ synchronized (changeThreadCount) {
+ final long cacheSizePerThread = threads.size() == 1 ?
0 : getCacheSizePerThread(threads.size() - 1);
+ resizeThreadCache(cacheSizePerThread);
threads.remove(streamThread);
-
resizeThreadCache(getCacheSizePerThread(threads.size()));
- return Optional.empty();
}
+ return Optional.of(streamThread.getName());
}
}
}
+ log.warn("Cannot remove a stream thread in state " + state());
Review comment:
There are two cases. Either there is no threads that can be removed of
the client is in the wrong state
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount + 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
+ TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
+
+
+ oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
Review comment:
yes
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
Review comment:
it can be either
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -924,25 +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()));
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
+ }
+ }
+ }
+ log.warn("Cannot add a stream thread in state " + state());
+ 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()) {
Review comment:
we also want to avoid removing threads in PENDING_SHUTDOWN. There should
not be any dead threads, this check is to make sure another call the
removeThread is not removing the same thread.
##########
File path: streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
##########
@@ -631,6 +634,23 @@ public void shouldNotAddThreadWhenError() {
assertThat(streams.threads.size(), equalTo(oldSize));
}
+ @Test
+ public void shouldRemoveThread() throws InterruptedException {
+ props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
+ final KafkaStreams streams = new
KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
+ streams.start();
+ final int oldSize = streams.threads.size();
+ TestUtils.waitForCondition(() -> streams.state() ==
KafkaStreams.State.RUNNING, 15L, "wait until running");
+ assertThat(streams.removeStreamThread(),
equalTo(Optional.of("newThread")));
+ assertThat(streams.threads.size(), equalTo(oldSize - 1));
+ }
+
+ @Test
+ public void shouldNotRemoveThreadWhenNotRunning() {
+ props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ final KafkaStreams streams = new
KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
+ assertThat(streams.removeStreamThread(), equalTo(Optional.empty()));
+ }
Review comment:
There is no case where a streamThread should be dead in the thread list
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount + 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
+ TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
+
+
+ oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"2", "3"}));
Review comment:
That is true, Will remove
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount + 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
+ TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
+
+
+ oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"2", "3"}));
+
+ final Optional<String> name2 = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
Review comment:
good catch
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount + 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
+ TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
+
+
+ oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"2", "3"}));
+
+ final Optional<String> name2 = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name2.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
Review comment:
will do
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -597,6 +598,18 @@ public void setStreamsUncaughtExceptionHandler(final
java.util.function.Consumer
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
}
+ public void waitOnThreadState(final StreamThread.State targetState) {
+ synchronized (stateLock) {
+ while (state != targetState) {
+ try {
+ stateLock.wait();
+ } catch (final InterruptedException e) {
+ e.printStackTrace();
Review comment:
I will do what the wait on state method in KafkaStreams does
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
Review comment:
we can verify here, we verify the names in a few lines as well
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
##########
@@ -108,4 +108,53 @@ public void shouldAddStreamThread() throws Exception {
TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
}
}
+
+ @Test
+ public void shouldRemoveStreamThread() throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ final int oldThreadCount =
kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount - 1));
+ }
+ }
+
+ @Test
+ public void shouldAddAndRemoveStreamThreadsWhileKeepingNamesCorrect()
throws Exception {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ int oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2"}));
+
+ final Optional<String> name = kafkaStreams.addStreamThread();
+
+ assertThat(name, CoreMatchers.not(Optional.empty()));
+ TestUtils.waitForCondition(
+ () -> kafkaStreams.localThreadsMetadata().stream().sequential()
+ .map(ThreadMetadata::threadName).anyMatch(t ->
t.equals(name.orElse(""))),
+ "Wait for the thread to be added"
+ );
+ assertThat(kafkaStreams.localThreadsMetadata().size(),
equalTo(oldThreadCount + 1));
+ assertThat(kafkaStreams.localThreadsMetadata().stream().map(t ->
t.threadName().split("-StreamThread-")[1]).sorted().toArray(), equalTo(new
String[] {"1", "2", "3"}));
+ TestUtils.waitForCondition(() -> kafkaStreams.state() ==
KafkaStreams.State.RUNNING, "wait for running");
+
+
+ oldThreadCount = kafkaStreams.localThreadsMetadata().size();
+ kafkaStreams.removeStreamThread();
Review comment:
yes
----------------------------------------------------------------
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
