You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/06/19 18:42:15 UTC
(pulsar) 01/07: [fix][client] Fix some potential resource leak (#24402)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dea3e57a6edb537df2d2a45c1da7bcd912fa73e8
Author: Xu Chengxin <57...@users.noreply.github.com>
AuthorDate: Wed Jun 11 00:50:12 2025 +0800
[fix][client] Fix some potential resource leak (#24402)
Fix resource leak using try-catch stmt method in #24389, and close other leaks in this project I've found later
(cherry picked from commit bfee7a68f84611b679a2d0e8942045462328f10a)
---
.../org/apache/pulsar/client/impl/AutoClusterFailover.java | 6 +++---
.../org/apache/pulsar/functions/runtime/RuntimeUtils.java | 14 +++++++-------
.../apache/pulsar/functions/worker/FunctionActioner.java | 12 +++++-------
.../pulsar/functions/worker/rest/api/ComponentImpl.java | 5 +++--
.../apache/pulsar/testclient/LoadSimulationController.java | 13 ++++++-------
5 files changed, 24 insertions(+), 26 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
index 68b781e67d2..844d1e2d253 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
@@ -128,9 +128,9 @@ public class AutoClusterFailover implements ServiceUrlProvider {
try {
resolver.updateServiceUrl(url);
InetSocketAddress endpoint = resolver.resolveHost();
- Socket socket = new Socket();
- socket.connect(new InetSocketAddress(endpoint.getHostName(), endpoint.getPort()), TIMEOUT);
- socket.close();
+ try (Socket socket = new Socket()) {
+ socket.connect(new InetSocketAddress(endpoint.getHostName(), endpoint.getPort()), TIMEOUT);
+ }
return true;
} catch (Exception e) {
log.warn("Failed to probe available, url: {}", url, e);
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
index 49a5dd40fa2..ae58fc40e2b 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
@@ -540,17 +540,17 @@ public class RuntimeUtils {
}
public static String getPrometheusMetrics(int metricsPort) throws IOException {
- StringBuilder result = new StringBuilder();
URL url = new URL(String.format("http://%s:%s", InetAddress.getLocalHost().getHostAddress(), metricsPort));
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod("GET");
- BufferedReader rd = new BufferedReader(new InputStreamReader(conn.getInputStream()));
- String line;
- while ((line = rd.readLine()) != null) {
- result.append(line + System.lineSeparator());
+ try (BufferedReader rd = new BufferedReader(new InputStreamReader(conn.getInputStream()))) {
+ StringBuilder result = new StringBuilder();
+ String line;
+ while ((line = rd.readLine()) != null) {
+ result.append(line + System.lineSeparator());
+ }
+ return result.toString();
}
- rd.close();
- return result.toString();
}
/**
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 389051fce42..4bdd120e215 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -266,13 +266,11 @@ public class FunctionActioner {
} else if (downloadFromPackageManagementService) {
getPulsarAdmin().packages().download(pkgLocationPath, tempPkgFile.getPath());
} else {
- FileOutputStream tempPkgFos = new FileOutputStream(tempPkgFile);
- WorkerUtils.downloadFromBookkeeper(
- dlogNamespace,
- tempPkgFos,
- pkgLocationPath);
- if (tempPkgFos != null) {
- tempPkgFos.close();
+ try (FileOutputStream tempPkgFos = new FileOutputStream(tempPkgFile)) {
+ WorkerUtils.downloadFromBookkeeper(
+ dlogNamespace,
+ tempPkgFos,
+ pkgLocationPath);
}
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
index ba87713d3c1..4058ebb667c 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
@@ -1283,8 +1283,9 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
if (worker().getWorkerConfig().isFunctionsWorkerEnablePackageManagement()) {
File tempFile = createPkgTempFile();
tempFile.deleteOnExit();
- FileOutputStream out = new FileOutputStream(tempFile);
- IOUtils.copy(uploadedInputStream, out);
+ try (FileOutputStream out = new FileOutputStream(tempFile)) {
+ IOUtils.copy(uploadedInputStream, out);
+ }
PackageMetadata metadata = PackageMetadata.builder().createTime(System.currentTimeMillis()).build();
worker().getBrokerAdmin().packages().upload(metadata, path, tempFile.getAbsolutePath());
} else {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java
index 2ba3eac171b..26f8cae012f 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java
@@ -638,14 +638,13 @@ public class LoadSimulationController extends CmdBase{
final List<String> commandArguments = arguments.commandArguments;
checkAppArgs(commandArguments.size() - 1, 1);
final String scriptName = commandArguments.get(1);
- final BufferedReader scriptReader = new BufferedReader(
- new InputStreamReader(new FileInputStream(Paths.get(scriptName).toFile())));
- String line = scriptReader.readLine();
- while (line != null) {
- read(line.split("\\s+"));
- line = scriptReader.readLine();
+ try (BufferedReader scriptReader = new BufferedReader(
+ new InputStreamReader(new FileInputStream(Paths.get(scriptName).toFile())))) {
+ String line;
+ while ((line = scriptReader.readLine()) != null) {
+ read(line.split("\\s+"));
+ }
}
- scriptReader.close();
break;
case "copy":
handleCopy(arguments);