func (p *UDFProcess) Close() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	err := p.server.Stop()
	p.processGroup.Wait()
	return err
}
Stop the UDFProcess cleanly.

Calling Close should only be done once the owner has stopped writing to the *In channel,
at which point the remaining data will be processed and the subprocess will be allowed to exit cleanly.
package kapacitor
import (
	"bufio"
	"fmt"
	"io"
	"log"
	"net"
	"sync"
	"time"
	"github.com/cenkalti/backoff"
	"github.com/influxdata/kapacitor/command"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/pipeline"
	"github.com/influxdata/kapacitor/udf"
	"github.com/influxdata/kapacitor/udf/agent"
	"github.com/pkg/errors"
)
type UDFNode struct {
	node
	u       *pipeline.UDFNode
	udf     udf.Interface
	aborted chan struct{}
	wg      sync.WaitGroup
	mu      sync.Mutex
	stopped bool
}
func newUDFNode(et *ExecutingTask, n *pipeline.UDFNode, l *log.Logger) (*UDFNode, error) {
	un := &UDFNode{
		node:    node{Node: n, et: et, logger: l},
		u:       n,
		aborted: make(chan struct{}),
	}
	f, err := et.tm.UDFService.Create(
		n.UDFName,
		et.Task.ID,
		n.Name(),
		l,
		un.abortedCallback,
	)
	if err != nil {
		return nil, err
	}
	un.udf = f
	un.node.runF = un.runUDF
	un.node.stopF = un.stopUDF
	return un, nil
}
var errNodeAborted = errors.New("node aborted")
func (n *UDFNode) stopUDF() {
	n.mu.Lock()
	defer n.mu.Unlock()
	if !n.stopped {
		n.stopped = true
		if n.udf != nil {
			n.udf.Abort(errNodeAborted)
		}
	}
}
func (n *UDFNode) runUDF(snapshot []byte) (err error) {
	defer func() {
		n.mu.Lock()
		defer n.mu.Unlock()
		if n.stopped && (err == udf.ErrServerStopped || err == errNodeAborted) {
			err = nil
		}
		n.stopped = true
	}()
	if err := n.udf.Open(); err != nil {
		return err
	}
	if err := n.udf.Init(n.u.Options); err != nil {
		return err
	}
	if snapshot != nil {
		if err := n.udf.Restore(snapshot); err != nil {
			return err
		}
	}
	forwardErr := make(chan error, 1)
	go func() {
		out := n.udf.Out()
		for m := range out {
			if err := edge.Forward(n.outs, m); err != nil {
				forwardErr <- err
				return
			}
		}
		forwardErr <- nil
	}()
	n.wg.Add(1)
	go func() {
		defer n.wg.Done()
		in := n.udf.In()
		for m, ok := n.ins[0].Emit(); ok; m, ok = n.ins[0].Emit() {
			n.timer.Start()
			select {
			case in <- m:
			case <-n.aborted:
				return
			}
			n.timer.Stop()
		}
	}()
	n.wg.Wait()
	if err := n.udf.Close(); err != nil {
		return err
	}
	return <-forwardErr
}
func (n *UDFNode) abortedCallback() {
	close(n.aborted)
	n.wg.Wait()
}
func (n *UDFNode) snapshot() ([]byte, error) {
	return n.udf.Snapshot()
}
type UDFProcess struct {
	taskName string
	nodeName string
	server    *udf.Server
	commander command.Commander
	cmdSpec   command.Spec
	cmd       command.Command
	stderr io.Reader
	processGroup   sync.WaitGroup
	logStdErrGroup sync.WaitGroup
	mu sync.Mutex
	logger        *log.Logger
	timeout       time.Duration
	abortCallback func()
}
func NewUDFProcess(
	taskName, nodeName string,
	commander command.Commander,
	cmdSpec command.Spec,
	l *log.Logger,
	timeout time.Duration,
	abortCallback func(),
) *UDFProcess {
	return &UDFProcess{
		taskName:      taskName,
		nodeName:      nodeName,
		commander:     commander,
		cmdSpec:       cmdSpec,
		logger:        l,
		timeout:       timeout,
		abortCallback: abortCallback,
	}
}
func (p *UDFProcess) Open() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	cmd := p.commander.NewCommand(p.cmdSpec)
	stdin, err := cmd.StdinPipe()
	if err != nil {
		return err
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return err
	}
	p.stderr = stderr
	err = cmd.Start()
	if err != nil {
		return err
	}
	p.cmd = cmd
	outBuf := bufio.NewReader(stdout)
	p.server = udf.NewServer(
		p.taskName,
		p.nodeName,
		outBuf,
		stdin,
		p.logger,
		p.timeout,
		p.abortCallback,
		cmd.Kill,
	)
	if err := p.server.Start(); err != nil {
		return err
	}
	p.logStdErrGroup.Add(1)
	go p.logStdErr()
	p.processGroup.Add(1)
	go func() {
		p.logStdErrGroup.Wait()
		p.server.WaitIO()
		err := cmd.Wait()
		if err != nil {
			err = fmt.Errorf("process exited unexpectedly: %v", err)
			defer p.server.Abort(err)
		}
		p.processGroup.Done()
	}()
	return nil
}