chore: Code cleanup of pipeline management (#3126)

* chore: Code cleanup of pipeline management * Fix checkstyle * Fix checkstyle * Fix pipeline update * Fix bug
apache · Aug 11, 2024 · 3ac47b9 · 3ac47b9
1 parent 225a213
commit 3ac47b9
Show file tree

Hide file tree

Showing 34 changed files with 68 additions and 794 deletions.
diff --git a/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingFormatException.java b/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingFormatException.java
diff --git a/...rc/main/java/org/apache/streampipes/commons/exceptions/NoMatchingJsonSchemaException.java b/...rc/main/java/org/apache/streampipes/commons/exceptions/NoMatchingJsonSchemaException.java
diff --git a/.../src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingProtocolException.java b/.../src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingProtocolException.java
diff --git a/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingSchemaException.java b/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoMatchingSchemaException.java
diff --git a/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoSepaInPipelineException.java b/...ns/src/main/java/org/apache/streampipes/commons/exceptions/NoSepaInPipelineException.java
diff --git a/...s/src/main/java/org/apache/streampipes/commons/exceptions/NoValidConnectionException.java b/...s/src/main/java/org/apache/streampipes/commons/exceptions/NoValidConnectionException.java
diff --git a/...mons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSecTypeException.java b/...mons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSecTypeException.java
diff --git a/...mons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSepTypeException.java b/...mons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSepTypeException.java
diff --git a/...rc/main/java/org/apache/streampipes/commons/exceptions/NoValidSepaStructureException.java b/...rc/main/java/org/apache/streampipes/commons/exceptions/NoValidSepaStructureException.java
diff --git a/...ons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSepaTypeException.java b/...ons/src/main/java/org/apache/streampipes/commons/exceptions/NoValidSepaTypeException.java
diff --git a/...in/java/org/apache/streampipes/commons/exceptions/RemoteServerNotAccessibleException.java b/...in/java/org/apache/streampipes/commons/exceptions/RemoteServerNotAccessibleException.java
diff --git a/...ommons/src/main/java/org/apache/streampipes/commons/exceptions/TooManyEdgesException.java b/...ommons/src/main/java/org/apache/streampipes/commons/exceptions/TooManyEdgesException.java
diff --git a/...in/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java b/...in/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java
@@ -19,8 +19,8 @@
 package org.apache.streampipes.connect.management.management;
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.manager.execution.PipelineExecutor;
 import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
-import org.apache.streampipes.manager.operations.Operations;
 import org.apache.streampipes.manager.pipeline.PipelineManager;
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.base.NamedStreamPipesEntity;
@@ -77,12 +77,12 @@ public void updateAdapter(AdapterDescription ad)
     affectedPipelines.forEach(p -> {
       var shouldRestartPipeline = p.isRunning();
       if (shouldRestartPipeline) {
-        Operations.stopPipeline(p, true);
+        new PipelineExecutor(p).stopPipeline(true);
       }
       var storedPipeline = PipelineManager.getPipeline(p.getPipelineId());
       var pipeline = applyUpdatedDataStream(storedPipeline, ad);
       try {
-        var modificationMessage = Operations.validatePipeline(pipeline);
+        var modificationMessage = new PipelineVerificationHandlerV2(pipeline).verifyPipeline();
         var updateInfo = makeUpdateInfo(modificationMessage, pipeline);
         var modifiedPipeline = new PipelineVerificationHandlerV2(pipeline).makeModifiedPipeline();
         var canAutoMigrate = canAutoMigrate(modificationMessage);
@@ -93,7 +93,7 @@ public void updateAdapter(AdapterDescription ad)
         }
         StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updateElement(modifiedPipeline);
         if (shouldRestartPipeline && canAutoMigrate) {
-          Operations.startPipeline(PipelineManager.getPipeline(p.getPipelineId()));
+          new PipelineExecutor(PipelineManager.getPipeline(p.getPipelineId())).startPipeline();
         }
       } catch (Exception e) {
         LOG.error("Could not update pipeline {}", pipeline.getName(), e);
@@ -113,7 +113,7 @@ public List<PipelineUpdateInfo> checkPipelineMigrations(AdapterDescription adapt
     affectedPipelines.forEach(pipeline -> {
       var updatedPipeline = applyUpdatedDataStream(pipeline, adapterDescription);
       try {
-        var modificationMessage = Operations.validatePipeline(updatedPipeline);
+        var modificationMessage = new PipelineVerificationHandlerV2(updatedPipeline).verifyPipeline();
         var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
         updateInfos.add(updateInfo);
       } catch (Exception e) {

diff --git a/...ne-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphHelpers.java b/...ne-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphHelpers.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.manager.data;
 
 import org.apache.streampipes.model.SpDataStream;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -30,10 +29,6 @@ public static List<SpDataStream> findStreams(PipelineGraph pipelineGraph) {
     return find(pipelineGraph, SpDataStream.class);
   }
 
-  public static List<InvocableStreamPipesEntity> findInvocableElements(PipelineGraph pipelineGraph) {
-    return find(pipelineGraph, InvocableStreamPipesEntity.class);
-  }
-
   private static <T> List<T> find(PipelineGraph pipelineGraph, Class<T> clazz) {
     return pipelineGraph
         .vertexSet()