Plugin initiated workflow might be stopped without panic

control/available_plugin.go

@@ -236,7 +236,7 @@ func (a *availablePlugin) Stop(r string) error {
 	return a.client.Kill(r)
 }
 
-// Kill assumes aplugin is not able to here a Kill RPC call
+// Kill assumes a plugin is not able to hear a Kill RPC call
 func (a *availablePlugin) Kill(r string) error {
 	log.WithFields(log.Fields{
 		"_module":     "control-aplugin",
@@ -252,6 +252,7 @@ func (a *availablePlugin) Kill(r string) error {
 		}).Debug("deleting available plugin package")
 		os.RemoveAll(filepath.Dir(a.execPath))
 	}
+
 	// If it's a streaming plugin, we need to signal the scheduler that
 	// this plugin is being killed.
 	if c, ok := a.client.(client.PluginStreamCollectorClient); ok {

control/plugin/client/grpc.go

@@ -219,10 +219,8 @@ func (g *grpcClient) Killed() {
 }
 
 func (g *grpcClient) Kill(reason string) error {
-
 	_, err := g.plugin.Kill(getContext(g.timeout), &rpc.KillArg{Reason: reason})
 	g.conn.Close()
-	g.Killed()
 	if err != nil {
 		return err
 	}

scheduler/task.go

@@ -295,8 +295,11 @@ func (t *task) stream() {
 			}
 			select {
 			case <-t.killChan:
+				t.Lock()
 				t.state = core.TaskStopped
-				break
+				t.Unlock()
+				done = true
+				return
 			case mts, ok := <-metricsChan:
 				if !ok {
 					metricsChan = nil