add serialize

riteshghorse · Sep 13, 2022 · 264c3fe · 264c3fe
1 parent 0275799
commit 264c3fe
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/serialize.go b/sdks/go/pkg/beam/core/runtime/graphx/serialize.go
@@ -28,6 +28,7 @@ import (
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
 	v1pb "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/v1"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/jsonx"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
@@ -520,6 +521,8 @@ func tryEncodeSpecial(t reflect.Type) (v1pb.Type_Special, bool) {
 		return v1pb.Type_BUNDLEFINALIZATION, true
 	case state.ProviderType:
 		return v1pb.Type_STATEPROVIDER, true
+	case timers.ProviderType:
+		return v1pb.Type_TIMERPROVIDER, true
 	case typex.KVType:
 		return v1pb.Type_KV, true
 	case typex.CoGBKType:
@@ -686,6 +689,8 @@ func decodeSpecial(s v1pb.Type_Special) (reflect.Type, error) {
 		return typex.BundleFinalizationType, nil
 	case v1pb.Type_STATEPROVIDER:
 		return state.ProviderType, nil
+	case v1pb.Type_TIMERPROVIDER:
+		return timers.ProviderType, nil
 	case v1pb.Type_KV:
 		return typex.KVType, nil
 	case v1pb.Type_COGBK: