Merge pull request #205 from grainier/fix-memory-issues

mohanvive · web-flow · commit a9da40559e23 · 2021-08-02T20:44:43.000+05:30
Fix executor service creation and memory releasing issue
diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java b/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java
@@ -319,6 +319,12 @@
                         type = {DataType.INT},
                         optional = true,
                         defaultValue = "100"),
+                @Parameter(
+                        name = "executor.service.threads",
+                        description = "Thread count for the executor service.",
+                        type = {DataType.INT},
+                        optional = true,
+                        defaultValue = "20"),
                 @Parameter(
                         name = "min.evictable.idle.time",
                         description = "Minimum time (in millis) a connection may sit idle in the " +
@@ -566,6 +572,8 @@ protected StateFactory init(StreamDefinition outputStreamDefinition, OptionHolde
         //pool configurations
         connectionPoolConfiguration = createPoolConfigurations(optionHolder);
 
+        executor = Executors.newFixedThreadPool(connectionPoolConfiguration.getExecutorServiceThreads());
+
         parametersList = optionHolder.validateAndGetStaticValue(HttpConstants.SINK_PARAMETERS, EMPTY_STRING);
 
         clientBootstrapConfiguration = optionHolder
@@ -1161,9 +1169,6 @@ public ClientConnector createClientConnector(DynamicOptions dynamicOptions) {
             senderConfig.disableSsl();
         }
 
-        executor = Executors.newFixedThreadPool(
-                senderConfig.getPoolConfiguration().getExecutorServiceThreads());
-
         //overwrite default transport configuration
         Map<String, Object> bootStrapProperties = HttpSinkUtil
                 .populateTransportConfiguration(clientBootstrapConfiguration);
diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/SSEWorkerThread.java b/component/src/main/java/io/siddhi/extension/io/http/sink/SSEWorkerThread.java
@@ -58,9 +58,10 @@ public void run() {
 
         try {
             buf.close();
-            carbonMessage.waitAndReleaseAllEntities();
         } catch (IOException e) {
             logger.error("Error occurred when closing the byte buffer in source " + streamID, e);
+        } finally {
+            carbonMessage.waitAndReleaseAllEntities();
         }
     }
 }
diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/WebSubHubSink.java b/component/src/main/java/io/siddhi/extension/io/http/sink/WebSubHubSink.java
@@ -299,6 +299,12 @@
                         type = {DataType.INT},
                         optional = true,
                         defaultValue = "100"),
+                @Parameter(
+                        name = "executor.service.threads",
+                        description = "Thread count for the executor service.",
+                        type = {DataType.INT},
+                        optional = true,
+                        defaultValue = "20"),
                 @Parameter(
                         name = "min.evictable.idle.time",
                         description = "Minimum time (in millis) a connection may sit idle in the " +
@@ -490,6 +496,7 @@ protected StateFactory init(StreamDefinition outputStreamDefinition, OptionHolde
 
         //pool configurations
         connectionPoolConfiguration = createPoolConfigurations(optionHolder);
+        executor = Executors.newFixedThreadPool(connectionPoolConfiguration.getExecutorServiceThreads());
         parametersList = optionHolder.validateAndGetStaticValue(HttpConstants.SINK_PARAMETERS, EMPTY_STRING);
         clientBootstrapConfiguration = optionHolder
                 .validateAndGetStaticValue(HttpConstants.CLIENT_BOOTSTRAP_CONFIGURATION, EMPTY_STRING);
@@ -760,9 +767,6 @@ public ClientConnector createClientConnector(String publisherURL) {
             senderConfig.disableSsl();
         }
 
-        executor = Executors.newFixedThreadPool(
-                senderConfig.getPoolConfiguration().getExecutorServiceThreads());
-
         //overwrite default transport configuration
         Map<String, Object> bootStrapProperties = HttpSinkUtil
                 .populateTransportConfiguration(clientBootstrapConfiguration);
@@ -823,7 +827,6 @@ private void setOnDemandQueryRuntimeForFindSubscription() {
                 siddhiAppContext, tableMap, windowMap, aggregationRuntimeMap);
     }
 
-
     public void setWebSubSubscriptionMap(Map<String, List<WebSubSubscriptionDTO>> webSubSubscriptionMap) {
         this.webSubSubscriptionMap = webSubSubscriptionMap;
     }
diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/util/HttpSinkUtil.java b/component/src/main/java/io/siddhi/extension/io/http/sink/util/HttpSinkUtil.java
@@ -295,6 +295,8 @@ private static Map<String, TrpPropertyTypes> trpPropertyTypeMap() {
     }
 
     public static PoolConfiguration createPoolConfigurations(OptionHolder optionHolder) {
+        int executorServiceThreads = Integer.parseInt(optionHolder.validateAndGetStaticValue(
+                HttpConstants.EXECUTOR_SERVICE_THREAD_COUNT, HttpConstants.DEFAULT_EXECUTOR_SERVICE_THREAD_COUNT));
         int maxIdlePerPool = Integer.parseInt(optionHolder.validateAndGetStaticValue(
                 HttpConstants.MAX_IDLE_CONNECTIONS_PER_POOL, HttpConstants.DEFAULT_MAX_IDLE_CONNECTIONS_PER_POOL));
         int minIdlePerPool = Integer.parseInt(optionHolder.validateAndGetStaticValue(
@@ -315,6 +317,7 @@ public static PoolConfiguration createPoolConfigurations(OptionHolder optionHold
         int maxWaitTime = Integer.parseInt(optionHolder.validateAndGetStaticValue(
                 HttpConstants.MAX_WAIT_TIME, HttpConstants.DEFAULT_MAX_WAIT_TIME));
         PoolConfiguration connectionPoolConfiguration = new PoolConfiguration();
+        connectionPoolConfiguration.setExecutorServiceThreads(executorServiceThreads);
         connectionPoolConfiguration.setMaxActivePerPool(maxActivePerPool);
         connectionPoolConfiguration.setMinIdlePerPool(minIdlePerPool);
         connectionPoolConfiguration.setMaxIdlePerPool(maxIdlePerPool);
diff --git a/component/src/main/java/io/siddhi/extension/io/http/source/HttpSyncWorkerThread.java b/component/src/main/java/io/siddhi/extension/io/http/source/HttpSyncWorkerThread.java
@@ -94,13 +94,13 @@ public void run() {
         } finally {
             try {
                 buf.close();
-                carbonMessage.waitAndReleaseAllEntities();
             } catch (IOException e) {
                 if (metrics != null) {
                     metrics.getTotalHttpErrorsMetric().inc();
                 }
-
                 logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);
+            } finally {
+                carbonMessage.waitAndReleaseAllEntities();
             }
         }
     }
diff --git a/component/src/main/java/io/siddhi/extension/io/http/source/HttpWebSubResponseProcessor.java b/component/src/main/java/io/siddhi/extension/io/http/source/HttpWebSubResponseProcessor.java
@@ -192,12 +192,13 @@ public void run() {
         } finally {
             try {
                 buf.close();
-                carbonMessage.waitAndReleaseAllEntities();
             } catch (IOException e) {
                 if (metrics != null) {
                     metrics.getTotalHttpErrorsMetric().inc();
                 }
                 logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);
+            } finally {
+                carbonMessage.waitAndReleaseAllEntities();
             }
         }
     }
diff --git a/component/src/main/java/io/siddhi/extension/io/http/source/HttpWorkerThread.java b/component/src/main/java/io/siddhi/extension/io/http/source/HttpWorkerThread.java
@@ -88,13 +88,13 @@ public void run() {
         } finally {
             try {
                 buf.close();
-                carbonMessage.waitAndReleaseAllEntities();
             } catch (IOException e) {
                 if (metrics != null) {
                     metrics.getTotalHttpErrorsMetric().inc();
                 }
-
                 logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);
+            } finally {
+                carbonMessage.waitAndReleaseAllEntities();
             }
         }
     }
diff --git a/component/src/main/java/io/siddhi/extension/io/http/util/HttpConstants.java b/component/src/main/java/io/siddhi/extension/io/http/util/HttpConstants.java
@@ -231,6 +231,8 @@ public class HttpConstants {
     public static final String BLOCKING_IO = "blocking.io";
 
     //pool configurations
+    public static final String EXECUTOR_SERVICE_THREAD_COUNT = "executor.service.threads";
+    public static final String DEFAULT_EXECUTOR_SERVICE_THREAD_COUNT = "20";
     public static final String MAX_ACTIVE_CONNECTIONS_PER_POOL = "max.pool.active.connections";
     public static final String DEFAULT_MAX_ACTIVE_CONNECTIONS_PER_POOL = "-1"; // unlimited
     public static final String MIN_IDLE_CONNECTIONS_PER_POOL = "min.pool.idle.connections";

Original file line number	Diff line number	Diff line change
`@@ -58,9 +58,10 @@ public void run() {`
`58`	`58`
`59`	`59`	`try {`
`60`	`60`	`buf.close();`
`61`		`- carbonMessage.waitAndReleaseAllEntities();`
`62`	`61`	`} catch (IOException e) {`
`63`	`62`	`logger.error("Error occurred when closing the byte buffer in source " + streamID, e);`
	`63`	`+ } finally {`
	`64`	`+ carbonMessage.waitAndReleaseAllEntities();`
`64`	`65`	`}`
`65`	`66`	`}`
`66`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -94,13 +94,13 @@ public void run() {`
`94`	`94`	`} finally {`
`95`	`95`	`try {`
`96`	`96`	`buf.close();`
`97`		`- carbonMessage.waitAndReleaseAllEntities();`
`98`	`97`	`} catch (IOException e) {`
`99`	`98`	`if (metrics != null) {`
`100`	`99`	`metrics.getTotalHttpErrorsMetric().inc();`
`101`	`100`	`}`
`102`		`-`
`103`	`101`	`logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);`
	`102`	`+ } finally {`
	`103`	`+ carbonMessage.waitAndReleaseAllEntities();`
`104`	`104`	`}`
`105`	`105`	`}`
`106`	`106`	`}`
Original file line number	Diff line number	Diff line change
`@@ -192,12 +192,13 @@ public void run() {`
`192`	`192`	`} finally {`
`193`	`193`	`try {`
`194`	`194`	`buf.close();`
`195`		`- carbonMessage.waitAndReleaseAllEntities();`
`196`	`195`	`} catch (IOException e) {`
`197`	`196`	`if (metrics != null) {`
`198`	`197`	`metrics.getTotalHttpErrorsMetric().inc();`
`199`	`198`	`}`
`200`	`199`	`logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);`
	`200`	`+ } finally {`
	`201`	`+ carbonMessage.waitAndReleaseAllEntities();`
`201`	`202`	`}`
`202`	`203`	`}`
`203`	`204`	`}`
Original file line number	Diff line number	Diff line change
`@@ -88,13 +88,13 @@ public void run() {`
`88`	`88`	`} finally {`
`89`	`89`	`try {`
`90`	`90`	`buf.close();`
`91`		`- carbonMessage.waitAndReleaseAllEntities();`
`92`	`91`	`} catch (IOException e) {`
`93`	`92`	`if (metrics != null) {`
`94`	`93`	`metrics.getTotalHttpErrorsMetric().inc();`
`95`	`94`	`}`
`96`		`-`
`97`	`95`	`logger.error("Error occurred when closing the byte buffer in source " + sourceID, e);`
	`96`	`+ } finally {`
	`97`	`+ carbonMessage.waitAndReleaseAllEntities();`
`98`	`98`	`}`
`99`	`99`	`}`
`100`	`100`	`}`