{"record":{"id":"b4fffaa1a0a59728","repo":"apache/beam","slug":"pubsubio-write-only-accepts-pcollections-of-v-and-v-received","errorCode":null,"errorMessage":"pubsubio.Write only accepts PCollections of %v and %v, received %v","messagePattern":"pubsubio\\.Write only accepts PCollections of (.+?) and (.+?), received (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/pubsubio/pubsubio.go","lineNumber":146,"sourceCode":"// Write writes PubSubMessages or []bytes to the given pubsub topic.\n// Panics if the input pcollection type is not one of those two types.\n//\n// When given []bytes, they are first wrapped in PubSubMessages.\n//\n// Note: Doesn't function in batch pipelines.\nfunc Write(s beam.Scope, project, topic string, col beam.PCollection) {\n\ts = s.Scope(\"pubsubio.Write\")\n\n\tpayload := &pipepb.PubSubWritePayload{\n\t\tTopic: pubsubx.MakeQualifiedTopicName(project, topic),\n\t}\n\n\tout := col\n\tif col.Type().Type() == reflectx.ByteSlice {\n\t\tout = beam.ParDo(s, wrapInMessage, col)\n\t}\n\tif out.Type().Type() != pubSubMessageT {\n\t\tpanic(fmt.Sprintf(\"pubsubio.Write only accepts PCollections of %v and %v, received %v\", pubSubMessageT, reflectx.ByteSlice, col.Type().Type()))\n\t}\n\tmarshaled := beam.ParDo(s, marshalMessageFn, out)\n\tbeam.External(s, writeURN, protox.MustEncode(payload), []beam.PCollection{marshaled}, nil, false)\n}\n","sourceCodeStart":128,"sourceCodeEnd":151,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/pubsubio/pubsubio.go#L128-L151","documentation":"pubsubio.Write only accepts PCollections whose element type is []byte or the internal PubSub message type. If the input PCollection has any other type after the automatic []byte wrapping step, the library panics because it cannot marshal the elements into Pub/Sub messages.","triggerScenarios":"Passing a PCollection of arbitrary structs, strings, or custom types directly to pubsubio.Write without first converting to []byte or the PubSub message type.","commonSituations":"Writing results of a ParDo that emit structs; forgetting to serialize with json.Marshal/protobuf before Write; changing an upstream PCollection's type and not updating the write path.","solutions":["Convert elements to []byte (e.g. json.Marshal or proto.Marshal) before pubsubio.Write.","Use the package's message wrapper type if you need attributes/ordering keys.","Add an assert on col.Type() in tests to catch type changes early."],"exampleFix":"// before\npubsubio.Write(s, project, topic, structCol)\n// after\nserialized := beam.ParDo(s, func(m MyMsg) []byte { b, _ := json.Marshal(m); return b }, structCol)\npubsubio.Write(s, project, topic, serialized)","handlingStrategy":"type-guard","validationCode":"if got := col.Type().Type(); got != reflectx.ByteSlice && got != pubSubMessageT {\n    return fmt.Errorf(\"pubsubio.Write needs []byte or PubSub message, got %v\", got)\n}","typeGuard":null,"tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if s, ok := r.(string); ok && strings.Contains(s, \"pubsubio.Write only accepts\") {\n            // handle\n        } else { panic(r) }\n    }\n}()","preventionTips":["Serialize elements to []byte (json/proto) before Write.","Pin the upstream PCollection type in tests so type changes fail fast.","Use the package's message type when attributes or ordering keys are needed."],"tags":["go","apache-beam","pubsub","panic","type-mismatch"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}