[fix][test] Stop worker services when tearing down PulsarFunctionTlsTest.
@@ -71,6 +71,7 @@ public class PulsarFunctionTlsTest {
protected PulsarService[] pulsarServices = new PulsarService[BROKER_COUNT];
protected PulsarService leaderPulsar;
protected PulsarAdmin leaderAdmin;
+ protected WorkerService[] fnWorkerServices = new WorkerService[BROKER_COUNT];
protected String testCluster = "my-cluster";
protected String testTenant = "my-tenant";
protected String testNamespace = testTenant + "/my-ns";
@@ -137,12 +138,12 @@ public class PulsarFunctionTlsTest {
workerConfig.setBrokerClientAuthenticationEnabled(true);
workerConfig.setTlsEnabled(true);
workerConfig.setUseTls(true);
- WorkerService fnWorkerService = WorkerServiceLoader.load(workerConfig);
+ fnWorkerServices[i] = WorkerServiceLoader.load(workerConfig);
configurations[i] = config;
pulsarServices[i] = new PulsarService(
- config, workerConfig, Optional.of(fnWorkerService), code -> {});
+ config, workerConfig, Optional.of(fnWorkerServices[i]), code -> {});
pulsarServices[i].start();
// Sleep until pulsarServices[0] becomes leader, this way we can spy namespace bundle assignment easily.
@@ -181,6 +182,9 @@ public class PulsarFunctionTlsTest {
if (pulsarAdmins[i] != null) {
pulsarAdmins[i].close();
}
+ if (fnWorkerServices[i] != null) {
+ fnWorkerServices[i].stop();
+ }
if (pulsarServices[i] != null) {
pulsarServices[i].close();
}
