Merge branch 'jg/optimizer_refactor' into master

Author: jgoldverg

helm/transfer-service/Charts.yaml

@@ -7,6 +7,10 @@
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.DefaultUriBuilderFactory;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 @Configuration
 public class OptimizerConfig {
 
@@ -19,4 +23,9 @@ public RestTemplate optimizerTemplate() {
                 .uriTemplateHandler(new DefaultUriBuilderFactory(optimizerUrl))
                 .build();
     }
+
+    @Bean(name ="optimizerTaskExecutor")
+    public Executor optimizerTaskExecutor(){
+        return Executors.newVirtualThreadPerTaskExecutor();
+    }
 }

src/main/java/org/onedatashare/transferservice/odstransferservice/config/OptimizerConfig.java

@@ -17,16 +17,13 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.JobParametersBuilder;
 import org.springframework.batch.core.launch.JobLauncher;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 
 @Service
 public class RabbitMQConsumer {
@@ -52,35 +49,38 @@ public RabbitMQConsumer(VfsExpander vfsExpander, Queue userQueue, JobParamServic
         this.jobLauncher = asyncJobLauncher;
         this.jc = jc;
         this.objectMapper = new ObjectMapper();
-        this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true);
         this.objectMapper.setDefaultPropertyInclusion(JsonInclude.Include.ALWAYS);
         this.threadPoolManager = threadPoolManager;
     }
 
     @RabbitListener(queues = "#{userQueue}")
     public void consumeDefaultMessage(final Message message) {
         String jsonStr = new String(message.getBody());
+
         logger.info("Message recv: {}", jsonStr);
         try {
             TransferJobRequest request = objectMapper.readValue(jsonStr, TransferJobRequest.class);
             logger.info("Job Recieved: {}", request.toString());
+
             if (request.getSource().getType().equals(EndpointType.vfs)) {
                 List<EntityInfo> fileExpandedList = vfsExpander.expandDirectory(request.getSource().getInfoList(), request.getSource().getFileSourcePath());
                 request.getSource().setInfoList(new ArrayList<>(fileExpandedList));
             }
             JobParameters parameters = jobParamService.translate(new JobParametersBuilder(), request);
             jc.setRequest(request);
             jobLauncher.run(jc.concurrentJobDefinition(), parameters);
+
             return;
         } catch (Exception e) {
-            logger.debug("Failed to parse jsonStr: {} to TransferJobRequest.java", jsonStr);
+            logger.error("Failed to parse jsonStr: {} to TransferJobRequest.java", jsonStr);
         }
         try {
             TransferApplicationParams params = objectMapper.readValue(jsonStr, TransferApplicationParams.class);
             logger.info("Parsed TransferApplicationParams: {}", params);
             this.threadPoolManager.applyOptimizer(params.getConcurrency(), params.getParallelism());
         } catch (Exception e) {
-            logger.info("Did not apply transfer params due to parsing message failure");
+            logger.error("Did not apply transfer params due to parsing message failure");
         }
     }
 }

src/main/java/org/onedatashare/transferservice/odstransferservice/consumer/RabbitMQConsumer.java

@@ -2,13 +2,15 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class EntityInfo {
     private String id;
     private String path;

src/main/java/org/onedatashare/transferservice/odstransferservice/model/EntityInfo.java

@@ -22,6 +22,7 @@ public class TransferJobRequest {
     private TransferOptions options;
     @JsonInclude(JsonInclude.Include.NON_NULL)
     private UUID jobUuid;
+    public String transferNodeName;
 
     @Data
     @AllArgsConstructor

src/main/java/org/onedatashare/transferservice/odstransferservice/model/TransferJobRequest.java

@@ -1,15 +1,19 @@
 package org.onedatashare.transferservice.odstransferservice.model.credential;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import lombok.Data;
 import lombok.ToString;
 import org.onedatashare.transferservice.odstransferservice.Enum.EndpointType;
 
 @Data
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class AccountEndpointCredential extends EndpointCredential{
     private String uri; //the hostname and port to reach the server
     private String username; //this should be the username for the client
     @ToString.Exclude
     private String secret; //This will contain the password of the resource you
+    @JsonIgnore
     byte[] encryptedSecret;
 
     public static String[] uriFormat(AccountEndpointCredential credential, EndpointType type) {

src/main/java/org/onedatashare/transferservice/odstransferservice/model/credential/AccountEndpointCredential.java

@@ -1,11 +1,13 @@
 package org.onedatashare.transferservice.odstransferservice.model.credential;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
 /**
  * Base class for storing one user credential
  */
 public class EndpointCredential{
-    protected String accountId;
 
+    String accountId;
     public EndpointCredential(){}
     public EndpointCredential(String accountId){
         this.accountId = accountId;

src/main/java/org/onedatashare/transferservice/odstransferservice/model/credential/EndpointCredential.java

@@ -1,22 +1,26 @@
 package org.onedatashare.transferservice.odstransferservice.model.optimizer;
 
+import lombok.AllArgsConstructor;
 import lombok.Data;
 
+import java.util.UUID;
+
 @Data
 public class OptimizerCreateRequest {
     String nodeId;
     int maxConcurrency;
     int maxParallelism;
     int maxPipelining;
     int maxChunkSize;
-
     String optimizerType;
-
     long fileCount;
     Long jobId;
     String dbType;
+    String jobUuid;
+    String userId;
 
-    public OptimizerCreateRequest(String nodeId, int maxConcurrency, int maxParallelism, int maxPipelining, String optimizerType, long fileCount, long jobId, String dbType) {
+    public OptimizerCreateRequest(String userId,String nodeId, int maxConcurrency, int maxParallelism, int maxPipelining, String optimizerType, long fileCount, long jobId, String dbType, String jobUuid) {
+        this.userId = userId;
         this.maxConcurrency = maxConcurrency;
         this.maxChunkSize = Integer.MAX_VALUE;
         this.maxParallelism = maxParallelism;
@@ -26,5 +30,6 @@ public OptimizerCreateRequest(String nodeId, int maxConcurrency, int maxParallel
         this.fileCount = fileCount;
         this.jobId = jobId;
         this.dbType = dbType;
+        this.jobUuid = jobUuid;
     }
 }

src/main/java/org/onedatashare/transferservice/odstransferservice/model/optimizer/OptimizerCreateRequest.java

@@ -66,19 +66,34 @@ public void applyOptimizer(int concurrency, int parallel) {
             if (key.contains(STEP_POOL_PREFIX)) {
                 logger.info("Changing {} pool size from {} to {}", pool.getThreadNamePrefix(), pool.getConcurrencyLimit(), concurrency);
                 if (concurrency > 0 && concurrency != pool.getConcurrencyLimit()) {
-//                    pool.setMaxPoolSize(concurrency);
                     pool.setConcurrencyLimit(concurrency);
                     logger.info("Set {} pool size to {}", pool.getThreadNamePrefix(), concurrency);
                 }
             }
+            if (key.contains(PARALLEL_POOL_PREFIX)) {
+                logger.info("Changing {} pool size from {} to {}", pool.getThreadNamePrefix(), pool.getConcurrencyLimit(), parallel);
+                if (parallel > 0 && parallel != pool.getConcurrencyLimit()) {
+                    pool.setConcurrencyLimit(parallel);
+                    logger.info("Set {} pool size to {}", pool.getThreadNamePrefix(), parallel);
+                }
+            }
         }
+
         for (String key : this.platformThreadMap.keySet()) {
             ThreadPoolTaskExecutor pool = this.platformThreadMap.get(key);
+            if (key.contains(STEP_POOL_PREFIX)) {
+                logger.info("Changing {} pool size from {} to {}", pool.getThreadNamePrefix(), pool.getCorePoolSize(), concurrency);
+                if (concurrency > 0 && concurrency != pool.getCorePoolSize()) {
+                    pool.setCorePoolSize(concurrency);
+                    logger.info("Set {} pool size to {}", pool.getThreadNamePrefix(), concurrency);
+                }
+            }
             if (key.contains(PARALLEL_POOL_PREFIX)) {
                 logger.info("Changing {} pool size from {} to {}", pool.getThreadNamePrefix(), pool.getCorePoolSize(), parallel);
-                if (parallel > 0 && parallel != pool.getPoolSize()) {
-//                    pool.setMaxPoolSize(parallel);
+                if (parallel > 0 && parallel != pool.getCorePoolSize()) {
                     pool.setCorePoolSize(parallel);
+                    logger.info("Set {} pool size to {}", pool.getThreadNamePrefix(), parallel);
+
                 }
             }
         }
@@ -100,7 +115,7 @@ public void clearJobPool() {
         logger.info("Cleared all thread pools");
     }
 
-//    public SimpleAsyncTaskExecutor sequentialThreadPool() {
+    //    public SimpleAsyncTaskExecutor sequentialThreadPool() {
 //        return this.createVirtualThreadExecutor(1, SEQUENTIAL_POOL_PREFIX);
 //    }
 //

src/main/java/org/onedatashare/transferservice/odstransferservice/pools/ThreadPoolManager.java

@@ -9,10 +9,13 @@
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;
 
+import java.util.concurrent.CompletableFuture;
+
 @Service
 public class OptimizerService {
 
@@ -31,14 +34,17 @@ public OptimizerService() {
         headers.setContentType(MediaType.APPLICATION_JSON);
     }
 
+    @Async("optimizerTaskExecutor")
     public void createOptimizerBlocking(OptimizerCreateRequest optimizerCreateRequest) throws RestClientException {
         optimizerCreateRequest.setNodeId(this.appName);
         logger.info("Sending OptimizerCreateRequest {}", optimizerCreateRequest);
         HttpEntity<OptimizerCreateRequest> createRequestHttpEntity = new HttpEntity<>(optimizerCreateRequest, this.headers);
         logger.info(createRequestHttpEntity.getBody().toString());
         this.optimizerTemplate.postForLocation("/optimizer/create", createRequestHttpEntity, Void.class);
+        CompletableFuture.completedFuture(null);
     }
 
+    @Async("optimizerTaskExecutor")
     public void deleteOptimizerBlocking(OptimizerDeleteRequest optimizerDeleteRequest) {
         optimizerDeleteRequest.setNodeId(this.appName);
         try {
@@ -47,5 +53,6 @@ public void deleteOptimizerBlocking(OptimizerDeleteRequest optimizerDeleteReques
             logger.error("Failed to Delete optimizer. {}", optimizerDeleteRequest);
         }
         logger.info("Deleted {}", optimizerDeleteRequest);
+        CompletableFuture.completedFuture(null);
     }
 }