apache/beam · error

panic(err)

Error message

panic(err)

What it means

schema.Transform builds a cross-language schema-transform call. It first converts the Go config value into an external configuration payload via xlangx.CreateExternalConfigurationPayload; if that conversion (reflection over the config struct into a schema + Row) fails, the panic propagates. Marshal errors for the SchemaTransformPayload also panic. It means your config value could not be represented as an external schema payload.

Solutions

1. Ensure the config is a plain struct with exported fields of schema-supported types (primitives, strings, bytes, nested structs, lists, maps with comparable keys).
2. Check for nil config; pass a zero-value struct instead of nil when no options are needed.
3. Compare your struct against the external transform's documented schema and add beam.RegisterSchemaProvider/type conversions for custom types.
4. Wrap construction in a helper that validates config fields before calling schema.Transform so failures surface early.

Example fix

// before
cfg := map[MyKey]string{...} // non-string map key: unsupported
schema.Transform(s, cfg, "beam:schemas:transform:v1")

// after
type Cfg struct {
    Name string `beam:"name"`
}
schema.Transform(s, Cfg{Name: "x"}, "beam:schemas:transform:v1")

func validateXlangConfig(cfg any) error {
    if cfg == nil {
        return errors.New("schema.Transform: config must not be nil")
    }
    rv := reflect.ValueOf(cfg)
    for rv.Kind() == reflect.Ptr {
        rv = rv.Elem()
    }
    if rv.Kind() != reflect.Struct {
        return fmt.Errorf("config must be a struct, got %T", cfg)
    }
    rt := rv.Type()
    for i := 0; i < rt.NumField(); i++ {
        if rt.Field(i).PkgPath != "" {
            return fmt.Errorf("field %s is unexported", rt.Field(i).Name)
        }
    }
    return nil
}

func isSchemaFriendly(cfg any) bool {
    if cfg == nil {
        return false
    }
    rv := reflect.ValueOf(cfg)
    for rv.Kind() == reflect.Ptr {
        rv = rv.Elem()
    }
    return rv.Kind() == reflect.Struct
}

func(cfg any, id string) (map[string]beam.PCollection, error) {
    defer func() {
        if r := recover(); r != nil {
            err = fmt.Errorf("xlang transform %s config rejected: %v", id, r)
        }
    }()
    out := schema.Transform(scope, cfg, id, opts...)
    return out, nil
}

When it happens

Trigger: Passing a config value that CreateExternalConfigurationPayload cannot encode (unsupported field types, unexported fields with no schema mapping, nil/incompatible config) to schema.Transform(scope, config, identifier, opts...).

Common situations: Hand-written config structs with unsupported types (maps with non-string keys, channels, funcs); passing nil config where a struct is required; refactoring an external transform's config struct so it no longer matches the external SDK's expected schema.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• tried converting invalid Node
• AfterProcessingTime trigger set without a delay or…
• At least one subtrigger required for composite triggers.
• attempted to add namespace to missing coder id
• attempted to add namespace to missing windowing strategy id

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/888f756211c60489. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/transforms/xlang/schema/external.go:76

// UnnamedOutputType specifies an output PCollection type of the transform.
// It must match the external transform's output schema.  This is simply
// syntactic sugar for OutputType(beam.UnnamedOutputTag(), tpe).
func UnnamedOutputType(tpe beam.FullType) Option {
	return OutputType(beam.UnnamedOutputTag(), tpe)
}

// ExpansionAddr is the URL of the expansion service to use.
func ExpansionAddr(addr string) Option {
	return func(opts *options) {
		opts.expansionAddr = addr
	}
}

// Transform configures a new cross language transform to call a "schema transform" in an external SDK.
func Transform(scope beam.Scope, config any, transformIdentifier string, opts ...Option) map[string]beam.PCollection {
	ecp, err := xlangx.CreateExternalConfigurationPayload(config)
	if err != nil {
		panic(err)
	}

	pl, err := proto.Marshal(&pipepb.SchemaTransformPayload{
		Identifier:          transformIdentifier,
		ConfigurationSchema: ecp.GetSchema(),
		ConfigurationRow:    ecp.GetPayload(),
	})
	if err != nil {
		panic(err)
	}

	allOpts := &options{}
	for _, o := range opts {
		o(allOpts)
	}

	return beam.CrossLanguage(scope, schemaTransformURN, pl, allOpts.expansionAddr, allOpts.inputs, allOpts.outputTypes)
}

View on GitHub (pinned to 12126d8942)