[incubator-streampipes] 03/03: fixed issue with offloading target selection

[go...@apache.org]

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

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

commit 4e57275424c2a11d9cad9514d878aa6fde903f15
Author: daniel-gomm <da...@outlook.de>
AuthorDate: Wed Oct 6 11:00:58 2021 +0200

    fixed issue with offloading target selection
---
 .../offloading/strategies/selection/RandomSelectionStrategy.java    | 6 +++++-
 .../streampipes/manager/migration/MigrationPipelineGenerator.java   | 2 ++
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/selection/RandomSelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/selection/RandomSelectionStrategy.java
index e1de78c..f89c9b5 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/selection/RandomSelectionStrategy.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/selection/RandomSelectionStrategy.java
@@ -19,16 +19,20 @@
 package org.apache.streampipes.node.controller.management.offloading.strategies.selection;
 
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
 import org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances;
 
 import java.util.List;
 import java.util.Random;
+import java.util.stream.Collectors;
 
 public class RandomSelectionStrategy implements SelectionStrategy{
 
     @Override
     public InvocableStreamPipesEntity select(List<InvocableStreamPipesEntity> blacklistedEntities) {
-        List<InvocableStreamPipesEntity> instances = RunningInvocableInstances.INSTANCE.getAll();
+        //Sinks cannot be migrated atm
+        List<InvocableStreamPipesEntity> instances = RunningInvocableInstances.INSTANCE.getAll().stream().
+                filter(e -> e instanceof DataProcessorInvocation).collect(Collectors.toList());
         if(instances.size() == 0)
             return null;
         return instances.get(new Random().nextInt(instances.size()));
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
index 586b89d..dba5c7b 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
@@ -144,6 +144,8 @@ public class MigrationPipelineGenerator {
                 .filter(dp -> dp.getDeploymentRunningInstanceId().equals(entityToMigrate.getDeploymentRunningInstanceId()))
                 .findFirst();
 
+        if (!originalInvocation.isPresent())
+            return null;
         int index = correspondingPipeline.getSepas().indexOf(originalInvocation.get());
 
         Pipeline targetPipeline;