Expose additional thread pool options (#1420)

Author: wivern

pom.xml

@@ -173,6 +173,9 @@
 
     <spring.batch.taskExecutor.corePoolSize>10</spring.batch.taskExecutor.corePoolSize>
     <spring.batch.taskExecutor.maxPoolSize>20</spring.batch.taskExecutor.maxPoolSize>
+    <spring.batch.taskExecutor.queueCapacity>2147483647</spring.batch.taskExecutor.queueCapacity>
+    <spring.batch.taskExecutor.threadGroupName></spring.batch.taskExecutor.threadGroupName>
+    <spring.batch.taskExecutor.threadNamePrefix></spring.batch.taskExecutor.threadNamePrefix>
 
     <!-- Sensitive Info settings -->
     <sensitiveinfo.admin.role>admin</sensitiveinfo.admin.role>

src/main/java/org/ohdsi/webapi/JobConfig.java

@@ -3,11 +3,13 @@
 import javax.annotation.PostConstruct;
 import javax.sql.DataSource;
 
+import org.apache.commons.lang3.StringUtils;
 import org.ohdsi.webapi.common.generation.AutoremoveJobListener;
 import org.ohdsi.webapi.common.generation.CancelJobListener;
 import org.ohdsi.webapi.job.JobTemplate;
 import org.ohdsi.webapi.service.JobService;
 import org.ohdsi.webapi.shiro.management.Security;
+import org.ohdsi.webapi.util.ManagedThreadPoolTaskExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.admin.service.*;
@@ -57,6 +59,15 @@ public class JobConfig {
 
     @Value("${spring.batch.taskExecutor.maxPoolSize}")
     private Integer maxPoolSize;
+
+    @Value("${spring.batch.taskExecutor.queueCapacity}")
+    private Integer queueCapacity;
+
+    @Value("${spring.batch.taskExecutor.threadGroupName}")
+    private String threadGroupName;
+
+    @Value("${spring.batch.taskExecutor.threadNamePrefix}")
+    private String threadNamePrefix;
 
     @Autowired
     private DataSource dataSource;
@@ -68,9 +79,16 @@ public String batchTablePrefix() {
 
     @Bean
     public TaskExecutor taskExecutor() {
-        final ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
+        final ThreadPoolTaskExecutor taskExecutor = new ManagedThreadPoolTaskExecutor();
         taskExecutor.setCorePoolSize(corePoolSize);
         taskExecutor.setMaxPoolSize(maxPoolSize);
+        taskExecutor.setQueueCapacity(queueCapacity);
+        if (StringUtils.isNotBlank(threadGroupName)) {
+            taskExecutor.setThreadGroupName(threadGroupName);
+        }
+        if (StringUtils.isNotBlank(threadNamePrefix)) {
+            taskExecutor.setThreadNamePrefix(threadNamePrefix);
+        }
         taskExecutor.afterPropertiesSet();
         return taskExecutor;
     }

src/main/java/org/ohdsi/webapi/util/ManagedThreadPoolTaskExecutor.java

@@ -0,0 +1,73 @@
+package org.ohdsi.webapi.util;
+
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+@ManagedResource
+public class ManagedThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
+
+  @ManagedAttribute
+  @Override
+  public int getCorePoolSize() {
+    return super.getCorePoolSize();
+  }
+
+  @ManagedAttribute
+  @Override
+  public int getMaxPoolSize() {
+    return super.getMaxPoolSize();
+  }
+
+  @ManagedAttribute
+  @Override
+  public int getKeepAliveSeconds() {
+    return super.getKeepAliveSeconds();
+  }
+
+  @ManagedAttribute
+  @Override
+  public int getPoolSize() {
+    return super.getPoolSize();
+  }
+
+  @ManagedAttribute
+  @Override
+  public int getActiveCount() {
+    return super.getActiveCount();
+  }
+
+  @ManagedAttribute
+  public long getCompletedTaskCount() {
+
+    Optional<ThreadPoolExecutor> executor = Optional.ofNullable(getThreadPoolExecutor());
+    return executor.map(ThreadPoolExecutor::getCompletedTaskCount).orElse(0L);
+  }
+
+  @ManagedAttribute
+  public int getLargestPoolSize() {
+
+    Optional<ThreadPoolExecutor> executor = Optional.ofNullable(getThreadPoolExecutor());
+    return executor.map(ThreadPoolExecutor::getLargestPoolSize).orElse(0);
+  }
+
+  @ManagedAttribute
+  public long getTaskCount() {
+
+    Optional<ThreadPoolExecutor> executor = Optional.ofNullable(getThreadPoolExecutor());
+    return executor.map(ThreadPoolExecutor::getTaskCount).orElse(0L);
+  }
+
+  @ManagedAttribute
+  public int getQueuedTaskCount() {
+
+    ThreadPoolExecutor executor = getThreadPoolExecutor();
+    BlockingQueue<Runnable> queue = Objects.nonNull(executor) ? executor.getQueue() : null;
+    return Objects.nonNull(queue) ? queue.size() : 0;
+  }
+}

src/main/resources/application.properties

@@ -69,6 +69,9 @@ spring.batch.repository.tableprefix=${spring.batch.repository.tableprefix}
 spring.batch.repository.isolationLevelForCreate=${spring.batch.repository.isolationLevelForCreate}
 spring.batch.taskExecutor.corePoolSize=${spring.batch.taskExecutor.corePoolSize}
 spring.batch.taskExecutor.maxPoolSize=${spring.batch.taskExecutor.maxPoolSize}
+spring.batch.taskExecutor.queueCapacity=${spring.batch.taskExecutor.queueCapacity}
+spring.batch.taskExecutor.threadGroupName=${spring.batch.taskExecutor.threadGroupName}
+spring.batch.taskExecutor.threadNamePrefix=${spring.batch.taskExecutor.threadNamePrefix}
 
 # EMBEDDED SERVER CONFIGURATION (ServerProperties)
 server.port = ${server.port}