Merge pull request #11 from RuiFG/master

fix: union operator rich can't be nil
RuiFG · Jan 9, 2023 · e70c5b2 · e70c5b2
2 parents 21e26f0 + 5996c4f
commit e70c5b2
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 4 deletions.
diff --git a/streaming-operator/union/union.go b/streaming-operator/union/union.go
@@ -39,15 +39,20 @@ func (o *operator[T]) NotifyCheckpointCancel(checkpointId int64) {
 }
 
 func (o *operator[T]) Open(ctx Context, emit element.Emit) error {
-	if err := o.rich.Open(ctx); err != nil {
-		return err
+	if o.rich != nil {
+		if err := o.rich.Open(ctx); err != nil {
+			return err
+		}
 	}
 	o.combineWatermark = NewCombineWatermark(o.inputCount)
 	return nil
 }
 
 func (o *operator[T]) Close() error {
-	return o.rich.Close()
+	if o.rich != nil {
+		return o.rich.Close()
+	}
+	return nil
 }
 
 func (o *operator[T]) ProcessElement(e element.Element, index int) {

diff --git a/streaming-operator/union/union_test.go b/streaming-operator/union/union_test.go