From 7c8bed8f5e8913684069f620846121d1f1940496 Mon Sep 17 00:00:00 2001 From: Vladislav Larkin Date: Fri, 7 Aug 2026 15:55:48 +0400 Subject: [PATCH 1/3] fix: stop the kafka watch thread when the client is closed KafkaMaaSClientImpl.close() interrupted the watchTopicCreate thread, but the thread only left its loop on an InterruptedException around wait(). While topic listeners were still registered, it sat in the inner polling loop, which swallows every exception and retries with no delay, so the thread outlived the client and kept hammering the agent. In tests this leaked thread flooded the MockServer shared by KafkaMaaSClientImplTest until it stopped responding, which failed unrelated tests in the class and blew prepare.log up to 569 MB during the monorepo release. The interrupt flag alone cannot drive the exit: the retry branch in HttpExecution calls Thread.sleep(), which clears the flag before the watch loop sees it. Track shutdown in a dedicated volatile field instead, and check it before the thread parks in wait(). KafkaMaaSClientCloseTest pins the invariant: once close() returns, the watch thread is gone. It uses its own lightweight HTTP stub rather than MockServer, so a regression cannot spill over into neighboring tests. --- .../impl/kafka/KafkaMaaSClientImpl.java | 10 +- .../impl/kafka/KafkaMaaSClientCloseTest.java | 139 ++++++++++++++++++ 2 files changed, 147 insertions(+), 2 deletions(-) create mode 100644 maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java index 26b3ff0de..4e9d0e62f 100644 --- a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java @@ -43,6 +43,7 @@ public class KafkaMaaSClientImpl implements KafkaMaaSClient { private final Duration watchTimeout = Duration.ofSeconds(60); // there is no need in highly concurrent map/lists implementation, we will wait for network responses most of the time private final Map>> topicCreateListeners = Collections.synchronizedMap(new HashMap<>()); + private volatile boolean closed = false; private final Lazy watchThread = new Lazy<>(() -> { Thread exec = new Thread(this::watchTenantCreateTopics, "watchTopicCreate"); exec.setDaemon(true); @@ -111,8 +112,8 @@ public void watchTenantTopics(String name, Consumer> callback private void watchTenantCreateTopics() { TypeReference> typeRef = new TypeReference<>() { }; - while (true) { - while (!topicCreateListeners.isEmpty()) { + while (!closed) { + while (!closed && !topicCreateListeners.isEmpty()) { String url = apiProvider.getKafkaTopicWatchCreateUrl(watchTimeout); List found = Collections.emptyList(); try { @@ -143,6 +144,10 @@ private void watchTenantCreateTopics() { } } + if (closed) { + return; + } + try { log.info("Nothing to watch, sleep thread."); synchronized (watchThread.get()) { @@ -223,6 +228,7 @@ public List search(SearchCriteria criteria) { @Override public void close() { + closed = true; if (watchThread.isInitialized()) { watchThread.get().interrupt(); try { diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java new file mode 100644 index 000000000..fe838381f --- /dev/null +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java @@ -0,0 +1,139 @@ +package com.netcracker.cloud.maas.client.impl.kafka; + +import static com.netcracker.cloud.maas.client.Utils.withProp; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.netcracker.cloud.maas.client.impl.ApiUrlProvider; +import com.netcracker.cloud.maas.client.impl.Env; +import com.netcracker.cloud.maas.client.impl.apiversion.ServerApiVersion; +import com.netcracker.cloud.maas.client.impl.http.HttpClient; +import com.netcracker.cloud.security.core.utils.k8s.M2MClientFactory; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +class KafkaMaaSClientCloseTest { + + private static final String WATCH_THREAD_NAME = "watchTopicCreate"; + private static final String WATCHED_TOPIC = "orders"; + private static final String NAMESPACE = "cloud-dev"; + private static final long LONG_POLL_DELAY_MILLIS = 200; + private static final String NO_TOPICS_YET = "[]"; + private static final String WATCHED_TOPIC_CREATED = "[{" + + "\"name\": \"maas.core-dev.orders\"," + + "\"classifier\": {\"name\": \"" + WATCHED_TOPIC + "\", \"namespace\": \"" + NAMESPACE + "\"}," + + "\"addresses\": {\"PLAINTEXT\": [\"localhost:9092\"]}" + + "}]"; + + private HttpServer agentStub; + private CountDownLatch watchPolled; + private CountDownLatch topicDelivered; + private final AtomicBoolean topicIsCreated = new AtomicBoolean(); + private Thread watchThread; + + @BeforeEach + void startAgentStub() throws IOException { + watchPolled = new CountDownLatch(1); + topicDelivered = new CountDownLatch(1); + topicIsCreated.set(false); + watchThread = null; + + agentStub = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + agentStub.createContext("/api-version", exchange -> respond(exchange, "{\"major\": 2, \"minor\": 8}")); + agentStub.createContext("/api/v2/kafka/topic/watch-create", this::answerWatchPoll); + agentStub.setExecutor(Executors.newCachedThreadPool()); + agentStub.start(); + } + + @AfterEach + void releaseWatchThreadAndStopStub() throws InterruptedException { + topicIsCreated.set(true); + if (watchThread != null && watchThread.isAlive()) { + topicDelivered.await(10, TimeUnit.SECONDS); + watchThread.interrupt(); + watchThread.join(TimeUnit.SECONDS.toMillis(5)); + } + agentStub.stop(0); + } + + @Test + void closeStopsWatchThread() { + withProp(Env.PROP_NAMESPACE, NAMESPACE, () -> { + String agentUrl = "http://localhost:" + agentStub.getAddress().getPort(); + withProp(Env.PROP_MAAS_AGENT_URL, agentUrl, () -> { + Set threadsBeforeWatch = Thread.getAllStackTraces().keySet(); + + KafkaMaaSClientImpl client = createKafkaClient(agentUrl); + client.watchTopicCreate(WATCHED_TOPIC, addr -> topicDelivered.countDown()); + + assertTrue(watchPolled.await(10, TimeUnit.SECONDS), + "the watch thread never reached the agent stub, so the client was not left watching"); + watchThread = findWatchThreadStartedAfter(threadsBeforeWatch); + assertNotNull(watchThread, "no new thread named '" + WATCH_THREAD_NAME + "' was started"); + + client.close(); + + assertFalse(watchThread.isAlive(), + "close() returned while '" + WATCH_THREAD_NAME + "' is still alive. " + + "The thread outlives the client and keeps polling " + agentUrl); + }); + }); + } + + private static Thread findWatchThreadStartedAfter(Set knownThreads) { + return Thread.getAllStackTraces().keySet().stream() + .filter(thread -> WATCH_THREAD_NAME.equals(thread.getName())) + .filter(thread -> !knownThreads.contains(thread)) + .findFirst() + .orElse(null); + } + + private static KafkaMaaSClientImpl createKafkaClient(String agentUrl) { + System.setProperty(M2MClientFactory.MAAS_AGENT_URL_PROP, agentUrl); + var httpClient = HttpClient.getMaasClient(() -> "faketoken"); + var serverApiVersion = new ServerApiVersion(httpClient, agentUrl); + System.clearProperty(M2MClientFactory.MAAS_AGENT_URL_PROP); + + return new KafkaMaaSClientImpl(httpClient, null, new ApiUrlProvider(serverApiVersion, agentUrl)); + } + + private void answerWatchPoll(HttpExchange exchange) throws IOException { + watchPolled.countDown(); + if (topicIsCreated.get()) { + respond(exchange, WATCHED_TOPIC_CREATED); + return; + } + try { + Thread.sleep(LONG_POLL_DELAY_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + respond(exchange, NO_TOPICS_YET); + } + + private static void respond(HttpExchange exchange, String body) throws IOException { + exchange.getRequestBody().readAllBytes(); + + byte[] payload = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, payload.length); + try (OutputStream response = exchange.getResponseBody()) { + response.write(payload); + } + } +} From 1527a710a1824480f333d91acc93e2360c8b24be Mon Sep 17 00:00:00 2001 From: Vladislav Larkin Date: Fri, 7 Aug 2026 16:20:48 +0400 Subject: [PATCH 2/3] test: trim the fixture in KafkaMaaSClientCloseTest JUnit builds a fresh instance per test method, so resetting the latches, the flag and the thread reference in the setup method was dead code. Initialize the fields inline instead. Drop the cached thread pool: the stub serves one client, and a user-supplied executor is not shut down by HttpServer.stop(), so it leaked threads on every run. --- .../impl/kafka/KafkaMaaSClientCloseTest.java | 21 ++++++------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java index fe838381f..45d1936b8 100644 --- a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java @@ -11,9 +11,7 @@ import java.nio.charset.StandardCharsets; import java.util.Set; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -33,36 +31,29 @@ class KafkaMaaSClientCloseTest { private static final String WATCHED_TOPIC = "orders"; private static final String NAMESPACE = "cloud-dev"; private static final long LONG_POLL_DELAY_MILLIS = 200; - private static final String NO_TOPICS_YET = "[]"; private static final String WATCHED_TOPIC_CREATED = "[{" + "\"name\": \"maas.core-dev.orders\"," + "\"classifier\": {\"name\": \"" + WATCHED_TOPIC + "\", \"namespace\": \"" + NAMESPACE + "\"}," + "\"addresses\": {\"PLAINTEXT\": [\"localhost:9092\"]}" + "}]"; + private final CountDownLatch watchPolled = new CountDownLatch(1); + private final CountDownLatch topicDelivered = new CountDownLatch(1); + private volatile boolean topicIsCreated; private HttpServer agentStub; - private CountDownLatch watchPolled; - private CountDownLatch topicDelivered; - private final AtomicBoolean topicIsCreated = new AtomicBoolean(); private Thread watchThread; @BeforeEach void startAgentStub() throws IOException { - watchPolled = new CountDownLatch(1); - topicDelivered = new CountDownLatch(1); - topicIsCreated.set(false); - watchThread = null; - agentStub = HttpServer.create(new InetSocketAddress("localhost", 0), 0); agentStub.createContext("/api-version", exchange -> respond(exchange, "{\"major\": 2, \"minor\": 8}")); agentStub.createContext("/api/v2/kafka/topic/watch-create", this::answerWatchPoll); - agentStub.setExecutor(Executors.newCachedThreadPool()); agentStub.start(); } @AfterEach void releaseWatchThreadAndStopStub() throws InterruptedException { - topicIsCreated.set(true); + topicIsCreated = true; if (watchThread != null && watchThread.isAlive()) { topicDelivered.await(10, TimeUnit.SECONDS); watchThread.interrupt(); @@ -114,7 +105,7 @@ private static KafkaMaaSClientImpl createKafkaClient(String agentUrl) { private void answerWatchPoll(HttpExchange exchange) throws IOException { watchPolled.countDown(); - if (topicIsCreated.get()) { + if (topicIsCreated) { respond(exchange, WATCHED_TOPIC_CREATED); return; } @@ -123,7 +114,7 @@ private void answerWatchPoll(HttpExchange exchange) throws IOException { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - respond(exchange, NO_TOPICS_YET); + respond(exchange, "[]"); } private static void respond(HttpExchange exchange, String body) throws IOException { From 5303ccca7f480329f485f33e657eec84fde506bd Mon Sep 17 00:00:00 2001 From: Vladislav Larkin Date: Fri, 7 Aug 2026 16:29:33 +0400 Subject: [PATCH 3/3] docs: state what KafkaMaaSClientCloseTest pins and why it avoids MockServer --- .../maas/client/impl/kafka/KafkaMaaSClientCloseTest.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java index 45d1936b8..9a1b28563 100644 --- a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientCloseTest.java @@ -25,6 +25,13 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpServer; +/** + * Pins the invariant that {@link KafkaMaaSClientImpl#close()} leaves no {@code watchTopicCreate} thread behind. + * + *

The stub server replaces MockServer on purpose: it answers the watch endpoint slowly and successfully, so a + * regression fails this assertion instead of flooding the MockServer instance that {@link KafkaMaaSClientImplTest} + * shares across its tests. + */ class KafkaMaaSClientCloseTest { private static final String WATCH_THREAD_NAME = "watchTopicCreate";