[BEAM-9615] Recover from registration panics. (apache#15228)

lostluck · web-flow · commit ccf55e759a3c · 2021-07-29T10:22:06.000-07:00
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go
@@ -106,7 +106,14 @@ func (r *Registry) RegisterType(ut reflect.Type) {
 }
 
 // reconcileRegistrations actually finishes the registration process.
-func (r *Registry) reconcileRegistrations() error {
+func (r *Registry) reconcileRegistrations() (deferedErr error) {
+	var ut reflect.Type
+	defer func() {
+		if r := recover(); r != nil {
+			deferedErr = errors.Errorf("panicked: %v", r)
+			deferedErr = errors.WithContextf(deferedErr, "reconciling schema registration for type %v", ut)
+		}
+	}()
 	for _, ut := range r.toReconcile {
 		check := func(ut reflect.Type) bool {
 			return coder.LookupCustomCoder(ut) != nil
