[incubator-streampipes] branch dev updated: [STREAMPIPES-248] Fix document conflict when restarting orphaned pipelines

[riemer]

This is an automated email from the ASF dual-hosted git repository.

riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git


The following commit(s) were added to refs/heads/dev by this push:
     new 1434f3f  [STREAMPIPES-248] Fix document conflict when restarting 
orphaned pipelines
1434f3f is described below

commit 1434f3ff195f4a829c4900a5b8f916ca761585e4
Author: Dominik Riemer <rie...@fzi.de>
AuthorDate: Tue May 18 21:00:06 2021 +0200

    [STREAMPIPES-248] Fix document conflict when restarting orphaned pipelines
---
 .../backend/StreamPipesBackendApplication.java          | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git 
a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
 
b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
index 53cd75b..790ee2b 100644
--- 
a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
+++ 
b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
@@ -24,6 +24,7 @@ import org.apache.streampipes.manager.operations.Operations;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
 import org.apache.streampipes.rest.notifications.NotificationListener;
+import org.apache.streampipes.storage.api.IPipelineStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -141,8 +142,9 @@ public class StreamPipesBackendApplication {
     PipelineOperationStatus status = Operations.startPipeline(pipeline);
     if (status.isSuccess()) {
       LOG.info("Pipeline {} successfully restarted", status.getPipelineName());
-      pipeline.setRestartOnSystemReboot(restartOnReboot);
-      
StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(pipeline);
+      Pipeline storedPipeline = 
getPipelineStorage().getPipeline(pipeline.getPipelineId());
+      storedPipeline.setRestartOnSystemReboot(restartOnReboot);
+      getPipelineStorage().updatePipeline(storedPipeline);
     } else {
       storeFailedRestartAttempt(pipeline);
       int failedAttemptCount = failedPipelines.get(pipeline.getPipelineId());
@@ -172,12 +174,17 @@ public class StreamPipesBackendApplication {
   }
 
   private List<Pipeline> getAllPipelines() {
-    return StorageDispatcher.INSTANCE
-            .getNoSqlStore()
-            .getPipelineStorageAPI()
+    return getPipelineStorage()
             .getAllPipelines();
   }
 
+  private IPipelineStorage getPipelineStorage() {
+    return StorageDispatcher
+            .INSTANCE
+            .getNoSqlStore()
+            .getPipelineStorageAPI();
+  }
+
   @Bean
   public FilterRegistrationBean shiroFilterBean() {
     FilterRegistrationBean<OncePerRequestFilter> bean = new 
FilterRegistrationBean<>();